Files d'attente de tâches en Go : création, traitement et optimisation

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

Dans le développement backend moderne, des tâches qui ne nécessitent pas de réponse immédiate à l'utilisateur apparaissent souvent : envoi d'email, génération de rapports, traitement d'images, synchronisation avec des API externes. Effectuer ces opérations de manière synchrone dans le cadre d'une requête HTTP est une mauvaise pratique, car cela ralentit la réponse et augmente la charge sur le serveur.

Les files d'attente de tâches (job queues) résolvent ce problème. En Go (Golang), les mécanismes de concurrence intégrés — goroutines et canaux — permettent de construire des files d'attente fiables et performantes sans dépendances externes. Dans cet article, nous allons voir comment créer une file d'attente de tâches à partir de zéro, l'optimiser avec un worker pool et éviter les erreurs typiques.

Fondamentaux : goroutines et canaux

Go fournit deux primitives clés pour travailler avec les files d'attente :

• Goroutines — des threads légers qui sont lancés avec le mot-clé go.
• Canaux — des « tuyaux » typés pour transférer des données entre les goroutines.

La file d'attente la plus simple est une goroutine qui lit depuis un canal et traite les tâches :

package main

import (    "fmt"    "time")

func worker(jobs <-chan int) {    for job := range jobs {        fmt.Printf("Traitement de la tâche %d", job)        time.Sleep(1 * time.Second) // simulation de travail    }}

func main() {    jobs := make(chan int, 10)

    // lancement du worker    go worker(jobs)

    // envoi des tâches    for i := 1; i <= 5; i++ {        jobs <- i    }

    close(jobs) // fermeture du canal après l'envoi de toutes les tâches    time.Sleep(2 * time.Second)}

Ce code crée un seul worker qui traite les tâches séquentiellement. Mais pour des projets réels, une évolutivité est nécessaire — plusieurs workers.

Implémentation d'un Worker Pool en Go

Un worker pool (pool de workers) est un modèle où nous lançons un nombre fixe de goroutines workers, chacune prenant des tâches depuis un canal commun. Cela permet de contrôler le degré de parallélisme et la charge sur le système.

Implémentation de base

package main

import (    "fmt"    "sync"    "time")

func worker(id int, jobs <-chan int, wg *sync.WaitGroup) {    defer wg.Done()    for job := range jobs {        fmt.Printf("Worker %d : tâche %d", id, job)        time.Sleep(500 * time.Millisecond)    }}

func main() {    const numWorkers = 3    const numJobs = 10

    jobs := make(chan int, numJobs)    var wg sync.WaitGroup

    // lancement des workers    for i := 1; i <= numWorkers; i++ {        wg.Add(1)        go worker(i, jobs, &wg)    }

    // envoi des tâches    for j := 1; j <= numJobs; j++ {        jobs <- j    }    close(jobs)

    wg.Wait() // attente de la fin de tous les workers    fmt.Println("Toutes les tâches sont terminées")}

Ici, nous utilisons sync.WaitGroup pour attendre la fin de tous les workers. Le canal avec un tampon numJobs permet d'envoyer toutes les tâches sans blocage, pendant que les workers ne les ont pas encore récupérées.

Ajout dynamique de tâches

Dans les applications réelles, les tâches arrivent de manière asynchrone — par exemple, via HTTP. Voici un exemple avec un gestionnaire :

package main

import (    "fmt"    "net/http"    "sync"    "time")

type Task struct {    ID   int    Data string}

var (    taskQueue = make(chan Task, 100)    wg        sync.WaitGroup)

func worker(id int) {    defer wg.Done()    for task := range taskQueue {        fmt.Printf("[Worker %d] Traitement : %s (ID=%d)", id, task.Data, task.ID)        time.Sleep(1 * time.Second)    }}

func handler(w http.ResponseWriter, r *http.Request) {    // récupération de la tâche depuis la requête (simplifié)    task := Task{ID: time.Now().Nanosecond(), Data: "Nouvelle tâche"}    taskQueue <- task    w.W