Motifs de concurrence Go : 7 motifs de production du Worker Pool au Pipeline

编程语言

Quand les fuites de goroutines rencontrent la concurrence non bornée : un cauchemar de production

3h du matin, alerte OOM en production. L'investigation révèle : un gestionnaire HTTP lançant des goroutines sans limite — 3 goroutines par requête, QPS 1000 signifie 3 000, et en 10 minutes, 180 000 goroutines fuites consomment toute la mémoire. Pire, ces goroutines détiennent des références à des canaux qui bloquent le GC, finissant par faire planter tout le nœud.

Ce n'est pas un cas isolé. Les primitives de concurrence de Go sont simples — go func() lance la concurrence en une ligne — mais la programmation concurrente en production est bien plus que simplement démarrer des goroutines. Vous devez contrôler les niveaux de concurrence, gérer la propagation d'erreurs, implémenter un arrêt propre, et prévenir les fuites de ressources. Cet article couvre 7 motifs de concurrence de niveau production pour vous aider à construire des services Go concurrents robustes.

Référence des concepts clés

Primitive Objectif Caractéristiques clés Cas d'usage typique
goroutine Unité d'exécution concurrente légère Pile extensible (2KB initial), ordonnancement par le runtime Go Toute tâche nécissant une exécution concurrente
channel Communication inter-goroutines Sûr pour les types, tamponnée/non tamponnée, fermable Passage de données, notification de signal, collecte de résultats
sync.WaitGroup Attendre un groupe de goroutines Trio Add/Done/Wait Attente de tâches par lot, fan-out concurent
sync.Mutex Exclusion mutuelle pour l'état partagé Utilisable à valeur zéro, support TryLock (Go 1.18+) Compteurs, mises à jour de cache, rechargement à chaud de config
context.Context Propager l'annulation et le timeout Immuable, ne peut dériver que des contextes enfant Timeout de requête, arrêt propre, traçage
errgroup.Group Exécution concurrente + collecte d'erreurs La première erreur annule toutes les goroutines Appels API par lot, récupération de données parallèle
semaphore.Weighted Limitation de débit par sémaphore pondéré Supporte les poids, acquisition consciente du timeout Limitation de débit API, contrôle de quota de ressources

5 défis de la programmation concurrente en production

Défi 1 : la concurrence non bornée provoque l'épuisement des ressources

func handleRequests(urls []string) {
    for _, url := range urls {
        go fetch(url)
    }
}

Chaque URL génère une goroutine — 10 000 URLs signifient 10 000 connexions concurrentes. Les pools de connexions de base de données s'épuisent, les services aval sont écrasés, la mémoire s'envole.

Défi 2 : fuites de goroutines

func process(ch <-chan int) {
    for {
        val := <-ch
        fmt.Println(val)
    }
}

Si le canal n'est jamais fermé, cette goroutine ne s'arrête jamais. Dans les services de longue durée, les goroutines fuites s'accumulent continuellement.

Défi 3 : erreurs silencieusement avalées

func fetchAll(urls []string) {
    var wg sync.WaitGroup
    for _, url := range urls {
        wg.Add(1)
        go func(u string) {
            defer wg.Done()
            resp, err := http.Get(u)
            if err != nil {
                log.Printf("fetch %s failed: %v", u, err)
                return
            }
            process(resp)
        }(url)
    }
    wg.Wait()
}

Les erreurs ne sont que journalisées — l'appelant n'a aucune idée des échecs. Si 2 URLs sur 3 échouent, l'appelant suppose que tout a réussi.

Défi 4 : pas d'arrêt propre (graceful shutdown)

Lorsque le service reçoit SIGTERM, les goroutines en cours sont interrompues de force. Les écritures en vol peuvent corrompre des données, les transactions en cours peuvent être à moitié terminées.

Défi 5 : difficulté de composer les motifs de concurrence

Le Worker Pool nécissite une limitation de débit, le Pipeline nécissite une propagation d'erreurs, le Fan-out nécissite une agrégation de résultats. Implémenter chaque motif seul n'est pas dur, mais les combiner dans un seul service crée des interactions qui mènent facilement aux interblocages (deadlocks).

7 motifs de concurrence de niveau production

Motif 1 : Worker Pool — pool de goroutines borné

Le Worker Pool est le motif de concurrence le plus fondamental et le plus important. Idée centrale : un nombre fixe d'ouvriers puisent des tâches dans une file, évitant la concurrence non bornée.

package workerpool

import (
    "context"
    "sync"
)

type Task func(ctx context.Context) error

type Pool struct {
    workers int
    tasks   chan Task
    wg      sync.WaitGroup
}

func NewPool(workers int, bufferSize int) *Pool {
    return &Pool{
        workers: workers,
        tasks:   make(chan Task, bufferSize),
    }
}

func (p *Pool) Start(ctx context.Context) {
    for i := 0; i < p.workers; i++ {
        p.wg.Add(1)
        go func(workerID int) {
            defer p.wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case task, ok := <-p.tasks:
                    if !ok {
                        return
                    }
                    _ = task(ctx)
                }
            }
        }(i)
    }
}

func (p *Pool) Submit(task Task) bool {
    select {
    case p.tasks <- task:
        return true
    default:
        return false
    }
}

func (p *Pool) Stop() {
    close(p.tasks)
    p.wg.Wait()
}

Exemple d'utilisation :

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    pool := NewPool(10, 100)
    pool.Start(ctx)

    urls := []string{
        "https://api.example.com/users",
        "https://api.example.com/orders",
        "https://api.example.com/products",
    }

    for _, url := range urls {
        u := url
        pool.Submit(func(ctx context.Context) error {
            return fetchURL(ctx, u)
        })
    }

    pool.Stop()
}

Points de conception clés :

  • Le nombre fixe d'ouvriers empêche l'explosion de goroutines
  • Le canal de tâches tamponné sert de file de tâches
  • Support du contexte pour l'annulation, les ouvriers peuvent sortir proprement
  • Submit est non bloquant, renvoie false quand la file est pleine

Motif 2 : Fan-out/Fan-in — diffusion-récolte parallèle

Le Fan-out distribue une source de données à plusieurs goroutines pour un traitement parallèle, le Fan-in fusionne les résultats de plusieurs goroutines dans un seul canal.

package fan

import (
    "context"
    "sync"
)

func FanOut[T any](ctx context.Context, source <-chan T, workers int) []<-chan T {
    channels := make([]<-chan T, workers)
    for i := 0; i < workers; i++ {
        ch := make(chan T)
        channels[i] = ch
        go func() {
            defer close(ch)
            for {
                select {
                case <-ctx.Done():
                    return
                case val, ok := <-source:
                    if !ok {
                        return
                    }
                    select {
                    case ch <- val:
                    case <-ctx.Done():
                        return
                    }
                }
            }
        }()
    }
    return channels
}

func FanIn[T any](ctx context.Context, channels ...<-chan T) <-chan T {
    out := make(chan T)
    var wg sync.WaitGroup
    wg.Add(len(channels))

    for _, ch := range channels {
        go func(c <-chan T) {
            defer wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case val, ok := <-c:
                    if !ok {
                        return
                    }
                    select {
                    case out <- val:
                    case <-ctx.Done():
                        return
                    }
                }
            }
        }(ch)
    }

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

    return out
}

Exemple d'utilisation — traitement parallèle de commandes :

func processOrders(ctx context.Context, orders <-chan Order) <-chan Result {
    workers := FanOut(ctx, orders, 5)
    return FanIn(ctx, workers...)
}

Points de conception clés :

  • Support des génériques (Go 1.18+), applicable à n'importe quel type
  • Chaque goroutine fan-out consomme indépendamment depuis le canal source
  • Le Fan-in utilise WaitGroup pour attendre la fermeture de tous les canaux d'entrée
  • Toutes les goroutines peuvent sortir à l'annulation du contexte

Motif 3 : Pipeline — traitement par étapes

Le Pipeline décompose un traitement complexe en étapes, chacune s'exécutant comme une goroutine reliée par des canaux.

package pipeline

import (
    "context"
)

type Stage[In any, Out any] func(ctx context.Context, in <-chan In) <-chan Out

func NewPipeline[In any, Out any](
    ctx context.Context,
    source <-chan In,
    stages ...Stage[In, In],
) <-chan Out {
    current := source
    for _, stage := range stages {
        current = stage(ctx, current)
    }
    return any(current).(<-chan Out)
}

func NewStage[In any, Out any](
    process func(ctx context.Context, in In) (Out, error),
    bufferSize int,
) Stage[In, Out] {
    return func(ctx context.Context, in <-chan In) <-chan Out {
        out := make(chan Out, bufferSize)
        go func() {
            defer close(out)
            for {
                select {
                case <-ctx.Done():
                    return
                case val, ok := <-in:
                    if !ok {
                        return
                    }
                    result, err := process(ctx, val)
                    if err != nil {
                        continue
                    }
                    select {
                    case out <- result:
                    case <-ctx.Done():
                        return
                    }
                }
            }
        }()
        return out
    }
}

Exemple d'utilisation — pipeline de traitement de données :

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    raw := make(chan RawData, 100)

    validate := NewStage[RawData, ValidData](func(ctx context.Context, in RawData) (ValidData, error) {
        if err := in.Validate(); err != nil {
            return ValidData{}, err
        }
        return in.ToValid(), nil
    }, 50)

    enrich := NewStage[ValidData, EnrichedData](func(ctx context.Context, in ValidData) (EnrichedData, error) {
        return fetchExtraInfo(ctx, in)
    }, 50)

    transform := NewStage[EnrichedData, FinalData](func(ctx context.Context, in EnrichedData) (FinalData, error) {
        return in.Transform()
    }, 50)

    result := NewPipeline(ctx, raw, validate, enrich, transform)

    go func() {
        for r := range result {
            saveToDB(r)
        }
    }()
}

Points de conception clés :

  • Chaque étape est une goroutine indépendante, peut être mise à l'échelle indépendamment
  • Les canaux fournissent une contre-pression (backpressure) entre les étapes
  • Les erreurs sont gérées dans les étapes, n'interrompent pas tout le pipeline
  • Toutes les étapes sortent proprement à l'annulation du contexte

Motif 4 : errgroup — gestion concurrente des erreurs

errgroup est la version consciente des erreurs de sync.WaitGroup : la première erreur annule toutes les goroutines.

package main

import (
    "context"
    "fmt"
    "net/http"
    "time"

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

type FetchResult struct {
    URL  string
    Body string
    Size int
}

func fetchMultiple(ctx context.Context, urls []string) ([]FetchResult, error) {
    g, ctx := errgroup.WithContext(ctx)
    results := make([]FetchResult, len(urls))

    for i, url := range urls {
        i, url := i, url
        g.Go(func() error {
            req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
            if err != nil {
                return fmt.Errorf("create request %s: %w", url, err)
            }

            resp, err := http.DefaultClient.Do(req)
            if err != nil {
                return fmt.Errorf("fetch %s: %w", url, err)
            }
            defer resp.Body.Close()

            if resp.StatusCode != http.StatusOK {
                return fmt.Errorf("fetch %s: status %d", url, resp.StatusCode)
            }

            body, err := io.ReadAll(resp.Body)
            if err != nil {
                return fmt.Errorf("read %s: %w", url, err)
            }

            results[i] = FetchResult{
                URL:  url,
                Body: string(body),
                Size: len(body),
            }
            return nil
        })
    }

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

errgroup avec limite de concurrence :

func fetchWithLimit(ctx context.Context, urls []string, maxConcurrent int) ([]FetchResult, error) {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(maxConcurrent)

    results := make([]FetchResult, len(urls))
    for i, url := range urls {
        i, url := i, url
        g.Go(func() error {
            results[i], _ = fetchOne(ctx, url)
            return nil
        })
    }

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

Points de conception clés :

  • errgroup.WithContext propage automatiquement l'annulation
  • g.SetLimit(n) contrôle la concurrence max (Go 1.20+)
  • La première erreur annule toutes les goroutines en cours
  • La capture de variable de clôture nécissite i, url := i, url

Motif 5 : Sémaphore — limitation de débit

semaphore.Weighted fournit des sémaphores pondérés pour l'allocation de ressources avec différents poids.

package ratelimit

import (
    "context"
    "fmt"

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

type RateLimiter struct {
    sem *semaphore.Weighted
}

func NewRateLimiter(maxWeight int64) *RateLimiter {
    return &RateLimiter{
        sem: semaphore.NewWeighted(maxWeight),
    }
}

func (r *RateLimiter) Do(ctx context.Context, weight int64, fn func() error) error {
    if err := r.sem.Acquire(ctx, weight); err != nil {
        return fmt.Errorf("acquire semaphore: %w", err)
    }
    defer r.sem.Release(weight)
    return fn()
}

Exemple d'utilisation — limitation de débit API étagée :

func main() {
    limiter := NewRateLimiter(100)

    err := limiter.Do(context.Background(), 10, func() error {
        return callLightAPI()
    })

    err = limiter.Do(context.Background(), 50, func() error {
        return callHeavyAPI()
    })
}

Sémaphore avec timeout :

func (r *RateLimiter) DoWithTimeout(ctx context.Context, weight int64, timeout time.Duration, fn func() error) error {
    acquireCtx, cancel := context.WithTimeout(ctx, timeout)
    defer cancel()

    if err := r.sem.Acquire(acquireCtx, weight); err != nil {
        return fmt.Errorf("semaphore acquire timeout: %w", err)
    }
    defer r.sem.Release(weight)
    return fn()
}

Points de conception clés :

  • Sémaphore pondéré, différentes opérations consomment différents quotas
  • Acquire supporte l'annulation de contexte et le timeout
  • Release doit être associé à Acquire
  • Adapté à la limitation de débit API étagée, au contrôle de quota de ressources

Motif 6 : annulation de contexte et timeout

Le contexte est le « fil vital » de la programmation concurrente Go, utilisé pour propager les signaux d'annulation, les timeouts et les échéances.

package ctxutil

import (
    "context"
    "fmt"
    "time"
)

type Result struct {
    Data  string
    Error error
}

func FetchWithTimeout(ctx context.Context, url string, timeout time.Duration) (string, error) {
    ctx, cancel := context.WithTimeout(ctx, timeout)
    defer cancel()

    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        return "", fmt.Errorf("create request: %w", err)
    }

    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        if ctx.Err() == context.DeadlineExceeded {
            return "", fmt.Errorf("fetch %s timed out after %v: %w", url, timeout, ctx.Err())
        }
        return "", fmt.Errorf("fetch %s: %w", url, err)
    }
    defer resp.Body.Close()

    body, err := io.ReadAll(resp.Body)
    if err != nil {
        return "", fmt.Errorf("read response: %w", err)
    }
    return string(body), nil
}

func BatchFetch(ctx context.Context, urls []string, timeout time.Duration) []Result {
    results := make([]Result, len(urls))
    var wg sync.WaitGroup

    for i, url := range urls {
        wg.Add(1)
        go func(idx int, u string) {
            defer wg.Done()
            data, err := FetchWithTimeout(ctx, u, timeout)
            results[idx] = Result{Data: data, Error: err}
        }(i, url)
    }

    wg.Wait()
    return results
}

Motif d'arrêt propre :

func main() {
    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer stop()

    server := &http.Server{Addr: ":8080"}

    go func() {
        <-ctx.Done()
        shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
        defer cancel()
        _ = server.Shutdown(shutdownCtx)
    }()

    _ = server.ListenAndServe()
}

Points de conception clés :

  • WithTimeout/WithDeadline pour définir des timeouts
  • WithCancel pour un contrôle d'annulation manuel
  • signal.NotifyContext pour écouter les signaux système
  • L'annulation de contexte se propage à toutes les goroutines enfant
  • defer cancel() prévient les fuites de contexte

Motif 7 : service concurrent de production — combinaison de motifs

Combinez les motifs ci-dessus pour construire un service concurrent de niveau production : Worker Pool + Pipeline + errgroup + Context.

package concurrencyservice

import (
    "context"
    "fmt"
    "sync"
    "time"

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

type Service struct {
    workers    int
    bufferSize int
    timeout    time.Duration
}

func NewService(workers, bufferSize int, timeout time.Duration) *Service {
    return &Service{
        workers:    workers,
        bufferSize: bufferSize,
        timeout:    timeout,
    }
}

type Job struct {
    ID    string
    Input any
}

type Output struct {
    Job    Job
    Result any
    Err    error
}

func (s *Service) Process(ctx context.Context, jobs []Job) ([]Output, error) {
    ctx, cancel := context.WithTimeout(ctx, s.timeout)
    defer cancel()

    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(s.workers)

    jobCh := make(chan Job, s.bufferSize)
    resultCh := make(chan Output, s.bufferSize)

    g.Go(func() error {
        defer close(jobCh)
        for _, job := range jobs {
            select {
            case <-ctx.Done():
                return ctx.Err()
            case jobCh <- job:
            }
        }
        return nil
    })

    var processWg sync.WaitGroup
    for i := 0; i < s.workers; i++ {
        processWg.Add(1)
        go func() {
            defer processWg.Done()
            for job := range jobCh {
                output := Output{Job: job}
                output.Result, output.Err = s.processOne(ctx, job)
                select {
                case resultCh <- output:
                case <-ctx.Done():
                    return
                }
            }
        }()
    }

    go func() {
        processWg.Wait()
        close(resultCh)
    }()

    var outputs []Output
    for result := range resultCh {
        outputs = append(outputs, result)
    }

    if err := g.Wait(); err != nil {
        return outputs, fmt.Errorf("service process: %w", err)
    }
    return outputs, nil
}

func (s *Service) processOne(ctx context.Context, job Job) (any, error) {
    select {
    case <-ctx.Done():
        return nil, ctx.Err()
    default:
    }
    return fmt.Sprintf("processed-%s", job.ID), nil
}

Points de conception clés :

  • errgroup contrôle la concurrence + la propagation d'erreurs
  • Canaux pour la distribution des tâches et la collecte des résultats
  • Contexte pour un timeout et une annulation unifiés
  • WaitGroup assure que tous les ouvriers terminent avant de fermer le canal de résultats
  • Conception en couches : ordonnancement (errgroup) + exécution (goroutines ouvrières) + collecte (range resultCh)

5 pièges courants et corrections

Piège 1 : capture de variable de clôture

❌ Incorrect :

for _, url := range urls {
    go func() {
        fetch(url)
    }()
}

Toutes les goroutines partagent la même variable url, finissant par récupérer la dernière URL.

✅ Correct :

for _, url := range urls {
    url := url
    go func() {
        fetch(url)
    }()
}

Piège 2 : canal non fermé provoque une fuite de goroutine

❌ Incorrect :

func producer() <-chan int {
    ch := make(chan int)
    go func() {
        for i := 0; i < 100; i++ {
            ch <- i
        }
    }()
    return ch
}

Les consommateurs parcourant le canal ne finissent jamais, fuite de goroutine.

✅ Correct :

func producer() <-chan int {
    ch := make(chan int)
    go func() {
        defer close(ch)
        for i := 0; i < 100; i++ {
            ch <- i
        }
    }()
    return ch
}

Piège 3 : position erreonée de WaitGroup Add

❌ Incorrect :

var wg sync.WaitGroup
for i := 0; i < 10; i++ {
    go func() {
        wg.Add(1)
        defer wg.Done()
        doWork()
    }()
}
wg.Wait()

Les goroutines n'ont peut-être pas encore atteint wg.Add(1), et wg.Wait() de la goroutine principale passe immédiatement.

✅ Correct :

var wg sync.WaitGroup
for i := 0; i < 10; i++ {
    wg.Add(1)
    go func() {
        defer wg.Done()
        doWork()
    }()
}
wg.Wait()

Piège 4 : fuite de time.After dans select

❌ Incorrect :

select {
case result := <-ch:
    process(result)
case <-time.After(5 * time.Second):
    return errors.New("timeout")
}

Chaque select crée un nouveau canal time.After. Des appels fréquents fuient des timers.

✅ Correct :

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

select {
case result := <-ch:
    process(result)
case <-ctx.Done():
    return ctx.Err()
}

Piège 5 : copie de Mutex

❌ Incorrect :

type SafeMap struct {
    mu   sync.Mutex
    data map[string]string
}

func copyMap(m SafeMap) SafeMap {
    return m
}

La valeur zéro de Mutex est « déverrouillée ». Copier un Mutex verrouillé provoque un interblocage ou une condition de concurrence.

✅ Correct :

type SafeMap struct {
    mu   *sync.Mutex
    data map[string]string
}

func NewSafeMap() *SafeMap {
    return &SafeMap{
        mu:   &sync.Mutex{},
        data: make(map[string]string),
    }
}

Référence de dépannage des erreurs

Symptôme Cause possible Investigation Solution
fatal error: all goroutines are asleep - deadlock! Toutes les goroutines bloquées, aucune goroutine active Vérifier l'appariement lecture/écriture de canal, select default Assurer que les canaux ont producteurs et consommateurs, ajouter l'annulation de contexte
Le nombre de goroutines ne cesse de croître Fuite de goroutine, canaux non fermés Monitoring runtime.NumGoroutine(), profil goroutine pprof Assurer que toutes les goroutines ont des chemins de sortie, defer close(ch)
La mémoire ne cesse de croître Les goroutines détiennent de grandes références d'objets, accumulation de canaux Profil heap pprof, vérifier la taille du tampon de canal Limiter le tampon de canal, libérer les références promptement
Erreurs context canceled fréquentes Contexte amont annulé, timeout trop court Vérifier la chaîne de contexte, vérifier que le timeout est raisonnable Ajuster le timeout, distinguer les erreurs métier des erreurs de timeout
Résultats concurrents manquants La goroutine sort avant d'écrire dans le canal Vérifier la logique de sortie de goroutine, confirmer la fermeture du canal Utiliser WaitGroup pour assurer que toutes les goroutines terminent avant de fermer le canal
race condition détectée Plusieurs goroutines lisent/écrivent des variables partagées go test -race, vérifier les variables globales et les captures de clôture Utiliser Mutex/RWMutex pour la protection, ou basculer vers la communication par canal
Canal bloqué, aucune réponse Déséquilibre producteur/consommateur Vérifier la taille du tampon de canal, monitor les taux de production/consommation Augmenter le tampon, utiliser select+default pour un envoi non bloquant
Échec de l'arrêt propre Les goroutines ne répondent pas à l'annulation de contexte Vérifier si les goroutines font select sur ctx.Done() Assurer que toutes les goroutines de longue durée vérifient ctx.Done()
errgroup ne renvoie qu'une seule erreur errgroup conçu pour l'annulation à première erreur Vérifier si vous devez collecter toutes les erreurs Utiliser un collecteur d'erreurs personnalisé ou plusieurs errgroups
Timeout d'acquisition de sémaphore Limite de concurrence trop stricte, requêtes en file trop longtemps Monitor le temps d'attente du sémaphore, ajuster les quotas de poids Augmenter la capacité du sémaphore, optimiser l'allocation des poids

Techniques d'optimisation avancées

Optimisation 1 : Worker Pool dynamique

Ajustez dynamiquement le nombre d'ouvriers selon la charge système :

package dynamicpool

import (
    "context"
    "runtime"
    "sync"
    "sync/atomic"
    "time"
)

type DynamicPool struct {
    minWorkers int64
    maxWorkers int64
    active     atomic.Int64
    tasks      chan Task
    wg         sync.WaitGroup
    adjustTick *time.Ticker
}

func NewDynamicPool(minW, maxW int) *DynamicPool {
    return &DynamicPool{
        minWorkers: int64(minW),
        maxWorkers: int64(maxW),
        tasks:      make(chan Task, maxW*2),
        adjustTick: time.NewTicker(5 * time.Second),
    }
}

func (p *DynamicPool) Start(ctx context.Context) {
    for i := 0; i < int(p.minWorkers); i++ {
        p.addWorker(ctx)
    }

    go func() {
        for {
            select {
            case <-ctx.Done():
                p.adjustTick.Stop()
                return
            case <-p.adjustTick.C:
                p.adjustWorkers(ctx)
            }
        }
    }()
}

func (p *DynamicPool) addWorker(ctx context.Context) {
    p.active.Add(1)
    p.wg.Add(1)
    go func() {
        defer func() {
            p.active.Add(-1)
            p.wg.Done()
        }()
        for task := range p.tasks {
            _ = task(ctx)
        }
    }()
}

func (p *DynamicPool) adjustWorkers(ctx context.Context) {
    current := p.active.Load()
    queueLen := len(p.tasks)

    if queueLen > int(current) && current < p.maxWorkers {
        p.addWorker(ctx)
    } else if queueLen == 0 && current > p.minWorkers {
        for i := 0; i < 5 && int(p.active.Load()) > int(p.minWorkers); i++ {
            p.tasks <- nil
        }
    }
}

func (p *DynamicPool) Stop() {
    close(p.tasks)
    p.wg.Wait()
}

Optimisation 2 : regroupement par lots (Batching)

Combinez plusieurs petites tâches en une seule opération par lot pour réduire les surcoûts syscall et IO :

package batcher

import (
    "context"
    "sync"
    "time"
)

type Batcher[T any, R any] struct {
    batchSize     int
    flushInterval time.Duration
    handler       func(ctx context.Context, batch []T) ([]R, error)
    mu            sync.Mutex
    buffer        []T
    results       []R
}

func NewBatcher[T any, R any](
    batchSize int,
    flushInterval time.Duration,
    handler func(ctx context.Context, batch []T) ([]R, error),
) *Batcher[T, R] {
    return &Batcher[T, R]{
        batchSize:     batchSize,
        flushInterval: flushInterval,
        handler:       handler,
        buffer:        make([]T, 0, batchSize),
    }
}

func (b *Batcher[T, R]) Add(ctx context.Context, item T) (R, error) {
    b.mu.Lock()
    b.buffer = append(b.buffer, item)

    if len(b.buffer) >= b.batchSize {
        batch := b.buffer
        b.buffer = make([]T, 0, b.batchSize)
        b.mu.Unlock()

        results, err := b.handler(ctx, batch)
        if err != nil {
            var zero R
            return zero, err
        }
        b.results = append(b.results, results...)
        var zero R
        if len(results) > 0 {
            return results[0], nil
        }
        return zero, nil
    }
    b.mu.Unlock()

    var zero R
    return zero, nil
}

func (b *Batcher[T, R]) Flush(ctx context.Context) ([]R, error) {
    b.mu.Lock()
    batch := b.buffer
    b.buffer = make([]T, 0, b.batchSize)
    b.mu.Unlock()

    if len(batch) == 0 {
        return b.results, nil
    }

    results, err := b.handler(ctx, batch)
    if err != nil {
        return nil, err
    }
    b.results = append(b.results, results...)
    return b.results, nil
}

Optimisation 3 : passage par canal sans copie (Zero-Copy)

Utilisez des pointeurs et sync.Pool pour réduire l'allocation mémoire lors du passage par canal :

package zerocopy

import (
    "sync"
)

type Buffer struct {
    Data []byte
}

var bufferPool = sync.Pool{
    New: func() any {
        return &Buffer{Data: make([]byte, 0, 4096)}
    },
}

func GetBuffer() *Buffer {
    return bufferPool.Get().(*Buffer)
}

func PutBuffer(buf *Buffer) {
    buf.Data = buf.Data[:0]
    bufferPool.Put(buf)
}

func ProcessPipeline(ctx context.Context, input <-chan []byte) <-chan *Buffer {
    out := make(chan *Buffer, 64)
    go func() {
        defer close(out)
        for {
            select {
            case <-ctx.Done():
                return
            case data, ok := <-input:
                if !ok {
                    return
                }
                buf := GetBuffer()
                buf.Data = append(buf.Data[:0], data...)

                select {
                case out <- buf:
                case <-ctx.Done():
                    PutBuffer(buf)
                    return
                }
            }
        }
    }()
    return out
}

Comparaison des motifs de concurrence

Caractéristique goroutine+channel sync.WaitGroup errgroup Worker Pool
Contrôle de concurrence Pas de contrôle intégré Pas de contrôle intégré SetLimit Nombre d'ouvriers fixe
Gestion d'erreurs Implémentation manuelle Implémentation manuelle Annulation auto à première erreur Implémentation manuelle
Contre-pression Tampon de canal Aucune Aucune Canal de tâches
Propagation d'annulation Contexte manuel Contexte manuel Contexte auto Contexte manuel
Collecte de résultats Réception de canal Variables partagées Retourner un slice Réception de canal
Cas d'usage Traitement de flux de données Attente de tâches par lot Appels API concurrents Traitement de tâches à débit limité
Complexité Moyenne Faible Faible Moyenne
Nombre de goroutines Non fixé Non fixé Limité Fixé
Recommandation production ⭐⭐⭐ ⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐

Résumé

La programmation concurrente Go ne consiste pas à « juste faire tourner » — elle consiste à répondre à quatre questions : Combien de goroutines tournent ? Quand sortent-elles ? Que se passe-t-il en cas d'erreur ? Comment les ressources sont-elles récupérées ? Le Worker Pool répond « combien », le Context répond « quand sortir », errgroup répond « que faire en erreur », et defer close() répond « comment récupérer ». Maîtrisez ces 7 motifs, et vous aurez la méthodologie centrale de la programmation concurrente Go de niveau production.

Outils recommandés

  • JSON Formatter — Formater les réponses JSON des services concurrents, déboguer rapidement les problèmes de structure de données
  • Hash Calculator — Calculer les signatures de requêtes et les checksums de données, assurer la cohérence des données entre requêtes concurrentes
  • Base64 Encode/Decode — Gérer l'encodage des données binaires dans les services concurrents

Essayez ces outils exécutés localement dans le navigateur — aucune inscription requise →

#Go#并发#goroutine#channel#sync#2026#并发模式