Files d'attente de tâches en Go : guide pratique

Introduction : pourquoi les files d'attente de tâches en Go ?

Dans le développement moderne, il est souvent nécessaire de traiter des tâches de manière asynchrone, sans bloquer le thread principal de l'application. Les files d'attente de tâches (task queues) sont un modèle classique qui permet de répartir le travail entre plusieurs gestionnaires (workers), de gérer la charge et d'assurer la tolérance aux pannes.

Le langage Go (Golang) fournit des mécanismes intégrés pour implémenter des files d'attente de tâches : les goroutines et les canaux (channels). Dans cet article, nous allons voir en pratique comment construire une file d'attente de tâches fiable avec la bibliothèque standard de Go, sans recourir à des courtiers de messages externes (bien qu'en production RabbitMQ ou Redis puissent être utiles).

Partie principale

1. Modèle de base du Worker Pool

Un Worker Pool est un ensemble d'un nombre fixe de goroutines (workers) qui écoutent un seul canal et traitent les tâches entrantes. Cela permet de contrôler le degré de parallélisme et d'éviter la surcharge du système.

Considérons un exemple simple : nous créons un pool de 3 workers qui traitent des nombres entiers (simulation de travail).

package main\n\nimport (\n    "fmt"\n    "sync"\n    "time"\n)\n\nfunc worker(id int, jobs <-chan int, wg *sync.WaitGroup) {\n    defer wg.Done()\n    for job := range jobs {\n        fmt.Printf("Worker %d a commencé la tâche %d\\n", id, job)\n        time.Sleep(time.Second) // simulation de travail\n        fmt.Printf("Worker %d a terminé la tâche %d\\n", id, job)\n    }\n}\n\nfunc main() {\n    const numJobs = 5\n    const numWorkers = 3\n\n    jobs := make(chan int, numJobs)\n    var wg sync.WaitGroup\n\n    // Lancement des workers\n    for w := 1; w <= numWorkers; w++ {\n        wg.Add(1)\n        go worker(w, jobs, &wg)\n    }\n\n    // Envoi des tâches\n    for j := 1; j <= numJobs; j++ {\n        jobs <- j\n    }\n    close(jobs) // fermeture du canal pour que les workers se terminent\n\n    wg.Wait()\n    fmt.Println("Toutes les tâches sont terminées")\n}

Analyse du code :

\n
• jobs — un canal tamponné dans lequel nous envoyons les tâches.
\n
• worker — une fonction qui lit depuis le canal et traite chaque tâche.
\n
• sync.WaitGroup — attend la fin de tous les workers.
\n
• Après l'envoi de toutes les tâches, le canal est fermé (close(jobs)), ce qui termine la boucle for range dans les workers.
\n

Ce modèle est la base de solutions plus complexes.

2. File d'attente avec priorités et timeouts

Dans les applications réelles, les tâches peuvent avoir des priorités différentes ou nécessiter une exécution dans un délai donné. Implémentons une file d'attente où les tâches prioritaires sont traitées en premier, et ajoutons un contexte avec timeout pour annuler les tâches longues.

package main\n\nimport (\n    "context"\n    "fmt"\n    "time"\n)\n\ntype Task struct {\n    ID       int\n    Priority int // plus le nombre est petit, plus la priorité est élevée\n    Data     string\n}\n\nfunc processTask(ctx context.Context, task Task) error {\n    select {\n    case <-ctx.Done():\n        return ctx.Err()\n    case <-time.After(time.Duration(task.Priority) * 500 * time.Millisecond):\n        fmt.Printf("Tâche %d traitée : %s (priorité %d)\\n", task.ID, task.Data, task.Priority)\n        return nil\n    }\n}\n\nfunc main() {\n    tasks := []Task{\n        {ID: 1, Priority: 3, Data: "Faible priorité"},\n        {ID: 2, Priority: 1, Data: "Haute priorité"},\n        {ID: 3, Priority: 2, Data: "Priorité moyenne"},\n    }\n\n    // Tri par priorité (les plus élevées en premier)\n    // Dans un projet réel, utilisez un tri ou un tas (container/heap)\n    for i := 0; i < len(tasks); i++ {\n        for j := i + 1; j < len(tasks); j++ {\n            if tasks[j].Priority < tasks[i].Priority {\n                tasks[i], tasks[j] = tasks[j], tasks[i]\n            }\n