Worker pool en Go : limiter les goroutines et annuler en toute sécurité

1. Aperçu

Les goroutines en Go sont peu coûteuses, donc appeler go pour chaque tâche est courant. Quand le nombre de tâches atteint des dizaines de milliers et que chaque tâche appelle une base de données ou HTTP, cette approche peut vider le connection pool ou dépasser le rate limit du service externe.

Cet article vous montre comment construire un worker pool, c’est-à-dire un nombre fixe de goroutines qui reçoivent du travail depuis un channel. À la fin, vous saurez comment écrire un pool basique, collecter les résultats et les erreurs, annuler avec context, simplifier avec errgroup.SetLimit, et tester que le pool limite vraiment le parallélisme.

2. Le problème : goroutines non limitées

Considérez ce code qui traite une liste de commandes :

for _, o := range orders {
	go chargeOrder(o) // 50 000 commandes -> 50 000 goroutines qui appellent l'API de paiement
}

Chaque goroutine ne consomme que quelques KB de mémoire, donc le problème se trouve rarement à ce niveau. Le problème vient des ressources en amont :

• Le pool de connexions de la base de données n’a que 20–50 connexions.
• La passerelle de paiement limite le nombre de requêtes par seconde.
• Le nombre de descripteurs de fichier du processus est limité.

Un worker pool résout ce problème en séparant deux concepts. Le nombre de tâches peut être très grand, tandis que le nombre de tâches s’exécutant simultanément est limité par le nombre de workers.

3. Worker pool basique

Le modèle comprend trois parties :

1. Un channel jobs qui contient le travail à faire.
2. N goroutines workers, chaque worker boucle avec for range sur le channel.
3. Un sync.WaitGroup pour attendre que tous les workers se terminent.

package main

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

type Order struct {
	ID     int
	Amount float64
}

func worker(id int, orders <-chan Order, wg *sync.WaitGroup) {
	defer wg.Done()
	for o := range orders {
		time.Sleep(100 * time.Millisecond) // simule l'appel à l'API de paiement
		fmt.Printf("worker %d traite la commande %d (%.0f)\n", id, o.ID, o.Amount)
	}
}

func main() {
	orders := make(chan Order)
	var wg sync.WaitGroup

	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go worker(i, orders, &wg)
	}

	start := time.Now()
	for i := 1; i <= 9; i++ {
		orders <- Order{ID: i, Amount: float64(i) * 10}
	}
	close(orders)
	wg.Wait()
	fmt.Println("terminé après", time.Since(start).Round(100*time.Millisecond))
}

Le code fonctionne comme suit :

• orders <-chan Order est un channel de réception du côté du worker. Le compilateur empêchera le worker d’envoyer ou de fermer le channel accidentellement.
• for o := range orders se ferme automatiquement quand le channel est fermé et vidé.
• close(orders) est le signal “fin du travail”. Si vous oubliez cette ligne, les workers attendent indéfiniment et wg.Wait() provoque un deadlock.
• wg.Add(1) doit être appelé avant go worker(...), pas à l’intérieur de la goroutine.

Résultat de l’exécution (l’ordre des lignes peut varier) :

worker 3 traite la commande 1 (10)
worker 1 traite la commande 2 (20)
worker 2 traite la commande 3 (30)
worker 2 traite la commande 4 (40)
...
terminé après 300ms

9 commandes divisées par 3 workers, chaque commande prenant 100ms, donc le temps total est d’environ 300ms au lieu de 900ms si exécuté séquentiellement.

Note : À partir de Go 1.25, vous pouvez écrire wg.Go(func() { worker(i, orders) }). Cette méthode appelle automatiquement Add et Done, évitant d’oublier l’un des deux.

4. Collecter les résultats et les erreurs

En pratique, vous avez souvent besoin de savoir quelle commande a réussi et laquelle a échoué. On ajoute un channel results pour cela :

package main

import (
	"errors"
	"fmt"
	"sync"
)

type Order struct {
	ID     int
	Amount float64
}

type Result struct {
	OrderID int
	Fee     float64
	Err     error
}

func chargeOrder(o Order) (float64, error) {
	if o.Amount <= 0 {
		return 0, errors.New("montant invalide")
	}
	return o.Amount * 0.02, nil // frais 2%
}

func main() {
	orders := []Order{{1, 100}, {2, 0}, {3, 250}, {4, 80}}

	jobs := make(chan Order)
	results := make(chan Result)
	var wg sync.WaitGroup

	for range 3 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for o := range jobs {
				fee, err := chargeOrder(o)
				results <- Result{OrderID: o.ID, Fee: fee, Err: err}
			}
		}()
	}

	go func() {
		for _, o := range orders {
			jobs <- o
		}
		close(jobs)
	}()

	go func() {
		wg.Wait()
		close(results)
	}()

	for r := range results {
		if r.Err != nil {
			fmt.Printf("commande %d erreur: %v\n", r.OrderID, r.Err)
			continue
		}
		fmt.Printf("commande %d frais %.2f\n", r.OrderID, r.Fee)
	}
}

Le résultat ressemble à :

commande 1 frais 2.00
commande 3 frais 5.00
commande 2 erreur: montant invalide
commande 4 frais 1.60

Il y a trois points à noter :

• L’envoi des jobs s’exécute dans une goroutine distincte. Si main envoie les jobs et lit les résultats sur la même goroutine, les workers seront bloqués à results <- tandis que main est bloqué à jobs <-. Cela provoque un deadlock.
• Une troisième goroutine attend wg.Wait() avant de fermer results. Seulement quand tous les workers ont terminé, on peut être sûr que personne n’envoie plus à results. Règle générale : seul l’émetteur ferme un channel.
• Les erreurs sont encapsulées dans Result plutôt que d’arrêter tout le pool. L’erreur d’une commande ne doit pas bloquer les autres.

L’ordre des résultats ne correspond pas à l’ordre des entrées. Si vous avez besoin de préserver l’ordre, transportez l’index avec la tâche, ou utilisez la méthode de la section 6.

5. Annuler le pool avec context

Quand une requête HTTP est annulée ou expire, le pool doit s’arrêter au lieu de continuer à appeler l’API de paiement. Vous le faites en laissant le worker écouter ctx.Done() :

func worker(ctx context.Context, jobs <-chan Order, results chan<- Result) {
	for {
		select {
		case <-ctx.Done():
			return
		case o, ok := <-jobs:
			if !ok {
				return
			}
			fee, err := chargeOrder(o)
			select {
			case results <- Result{OrderID: o.ID, Fee: fee, Err: err}:
			case <-ctx.Done():
				return
			}
		}
	}
}

Le côté producteur a également besoin d’un select similaire :

go func() {
	defer close(jobs)
	for _, o := range orders {
		select {
		case jobs <- o:
		case <-ctx.Done():
			return
		}
	}
}()

Et créer un context avec timeout à l’endroit de l’appel :

ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()

Pourquoi select dans les deux directions d’envoi : si un worker ne vérifie ctx que lors de la réception d’une tâche, il peut rester bloqué indéfiniment à results <- si le lecteur a quitté. C’est une source très commune de fuite de goroutines.

Note : select aide le worker à quitter entre les tâches. Si chargeOrder effectue un appel HTTP, vous devez passer ctx à l’intérieur (http.NewRequestWithContext) pour annuler la requête en cours.

6. Simplifier avec errgroup.SetLimit

Écrire manuellement les channels, WaitGroup et context est sujet aux erreurs. Le package errgroup a SetLimit, qui permet de créer un pool limité en quelques lignes. Tout d’abord, ajoutez la dépendance :

go get golang.org/x/sync/errgroup

La fonction générique ci-dessous traite une slice avec un maximum de limit goroutines, préserve l’ordre des résultats et s’arrête à la première erreur :

package pool

import (
	"context"

	"golang.org/x/sync/errgroup"
)

func Process[T, R any](
	ctx context.Context,
	limit int,
	items []T,
	fn func(context.Context, T) (R, error),
) ([]R, error) {
	results := make([]R, len(items))
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(limit)

	for i, item := range items {
		g.Go(func() error {
			r, err := fn(ctx, item)
			if err != nil {
				return err
			}
			results[i] = r
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err
	}
	return results, nil
}

Le code fonctionne comme suit :

• g.SetLimit(limit) rend g.Go bloquant quand il y a déjà limit goroutines en cours. En termes d’efficacité, c’est équivalent à un worker pool.
• errgroup.WithContext retourne un ctx qui sera annulé dès qu’une fonction retourne une erreur. Les autres tâches utilisant ctx recevront le signal d’arrêt.
• results[i] = r est sûr lors d’exécutions concurrentes car chaque goroutine écrit à un index distincte. Pas besoin de mutex dans ce cas.
• À partir de Go 1.22, les variables de boucle i et item sont créées à nouveau à chaque itération, donc la closure n’utilise pas la même variable que dans les versions antérieures.

Comment l’appeler :

fees, err := pool.Process(ctx, 5, orders, func(ctx context.Context, o Order) (float64, error) {
	return paymentService.Charge(ctx, o)
})

Différence avec un pool écrit manuellement : errgroup crée une nouvelle goroutine pour chaque item au lieu de réutiliser N workers. Pour la plupart des applications, cette différence est négligeable. Si vous avez besoin que les workers maintiennent un état, par exemple chaque worker ayant une connexion, utilisez le pool écrit manuellement des sections 3–5.

7. Tester le worker pool

Vous devez tester deux comportements principaux. D’abord, le pool ne dépasse jamais la limit. Deuxièmement, les erreurs sont retournées correctement. Pour mesurer le niveau de parallélisme, on utilise atomic.Int32 :

package pool

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"
)

func TestProcess_LimitsConcurrencyAndKeepsOrder(t *testing.T) {
	var running, maxSeen atomic.Int32
	items := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}

	got, err := Process(context.Background(), 3, items, func(_ context.Context, n int) (int, error) {
		cur := running.Add(1)
		defer running.Add(-1)
		for {
			m := maxSeen.Load()
			if cur <= m || maxSeen.CompareAndSwap(m, cur) {
				break
			}
		}
		time.Sleep(10 * time.Millisecond)
		return n * 2, nil
	})

	if err != nil {
		t.Fatalf("erreur inattendue: %v", err)
	}
	if maxSeen.Load() > 3 {
		t.Errorf("concurrence %d, limite 3", maxSeen.Load())
	}
	want := []int{2, 4, 6, 8, 10, 12, 14, 16, 18, 20}
	for i := range want {
		if got[i] != want[i] {
			t.Fatalf("got[%d] = %d, want %d", i, got[i], want[i])
		}
	}
}

func TestProcess_ReturnsFirstError(t *testing.T) {
	errDeclined := errors.New("carte refusée")

	_, err := Process(context.Background(), 2, []int{1, 2, 3}, func(_ context.Context, n int) (int, error) {
		if n == 2 {
			return 0, errDeclined
		}
		return n, nil
	})

	if !errors.Is(err, errDeclined) {
		t.Fatalf("attendu errDeclined, reçu %v", err)
	}
}

Exécutez le test avec le détecteur de races :

go test -race ./...

ok  	example.com/shop/pool	0.061s

La boucle CompareAndSwap met à jour maxSeen sans verrou. Le flag -race est obligatoire pour le code concurrent, car il détecte les accès de données partagées que l’œil nu peut facilement manquer.

8. Erreurs courantes et quand ne pas utiliser

Erreurs courantes :

• Oublier close(jobs) : les workers ne terminent jamais et le programme signale fatal error: all goroutines are asleep - deadlock!.
• Fermer le channel du côté du récepteur, ou deux fois : provoque panic: close of closed channel.
• Appeler wg.Add à l’intérieur de la goroutine : Wait peut retourner avant que la goroutine n’ait le temps d’appeler Add.
• Envoyer des jobs et lire les résultats sur la même goroutine avec un channel non bufférisé : provoque un deadlock comme analysé à la section 4.
• Ne pas faire select sur ctx.Done() lors de l’envoi des résultats : entraîne une fuite de goroutines si le lecteur a quitté plus tôt.
• Choisir arbitrairement le nombre de workers : pour les tâches CPU-bound, utilisez runtime.NumCPU(). Pour les tâches I/O-bound, basez-vous sur la limite des ressources en amont, par exemple la taille du pool de connexions.

Quand ne pas utiliser un worker pool :

• Le nombre de tâches est petit et connu à l’avance, par exemple appeler 3 services en parallèle. Dans ce cas, errgroup sans SetLimit est suffisant.
• Les tâches sont extrêmement courtes, seulement quelques microsecondes. Le coût de passage par le channel peut être plus grand que le travail lui-même, donc traiter séquentiellement ou par lots sera plus efficace.
• Vous devez limiter selon le débit (requêtes/seconde) et non le nombre simultané. Dans ce cas, vous avez besoin d’un rate limiter, par exemple golang.org/x/time/rate. Un worker pool ne peut pas remplacer un rate limiter.

9. Conclusion

Un worker pool vous aide à contrôler le niveau de parallélisme en Go : le nombre de tâches peut être très grand mais les ressources en amont sont toujours protégées. L’article a couvert les étapes suivantes :

• Pool basique avec channel et sync.WaitGroup.
• Collecter les résultats et erreurs via le channel results, fermer le channel correctement.
• Annulation sécurisée avec context dans les deux directions d’envoi.
• Simplification avec errgroup.SetLimit en utilisant une fonction Process générique.
• Test de la limite de concurrence et des erreurs avec go test -race.

Dans la plupart des cas, errgroup.SetLimit est le choix le plus simple et le moins sujet aux erreurs. Vous ne devriez écrire votre propre pool que si les workers doivent maintenir un état ou si vous avez besoin de contrôler finement le flux des tâches et des résultats.