Patrones de Concurrencia en Go: 7 Patrones de Producción desde Worker Pool hasta Pipeline

编程语言

Cuando las Fugas de Goroutines se Encuentran con Concurrencia Sin Límites: Una Pesadilla en Producción

3 AM, alerta de OOM en producción. La investigación revela: un handler HTTP lanzando goroutines sin límites — 3 goroutines por petición, QPS 1000 significa 3,000, y en 10 minutos, 180,000 goroutines filtradas consumen toda la memoria. Peor aún, estas goroutines mantienen referencias de canales que bloquean el GC, eventualmente colapsando todo el nodo.

Este no es un caso aislado. Las primitivas de concurrencia de Go son simples — go func() lanza concurrencia en una línea — pero la programación concurrente en producción es mucho más que simplemente iniciar goroutines. Necesitas controlar los niveles de concurrencia, manejar la propagación de errores, implementar apagado graceful y prevenir fugas de recursos. Este artículo cubre 7 patrones de concurrencia de grado de producción para ayudarte a construir servicios concurrentes robustos en Go.

Referencia de Conceptos Fundamentales

Primitiva Propósito Características Clave Caso de Uso Típico
goroutine Unidad de ejecución concurrente ligera Pila creciente (2KB inicial), planificación del runtime de Go Cualquier tarea que requiera ejecución concurrente
channel Comunicación entre goroutines Seguro por tipo, con/sin búfer, cerrable Paso de datos, notificación de señales, recolección de resultados
sync.WaitGroup Esperar un grupo de goroutines Trío Add/Done/Wait Espera de tareas por lotes, fan-out concurrente
sync.Mutex Exclusión mutua para estado compartido Utilizable con valor cero, soporte TryLock (Go 1.18+) Contadores, actualizaciones de caché, hot-reload de configuración
context.Context Propagar cancelación y timeout Inmutable, solo se pueden derivar contextos hijos Timeout de peticiones, apagado graceful, tracing
errgroup.Group Ejecución concurrente + recolección de errores El primer error cancela todas las goroutines Llamadas API por lotes, obtención paralela de datos
semaphore.Weighted Limitación de tasa con semáforo ponderado Soporta pesos, adquisición consciente de timeout Limitación de tasa de API, control de cuota de recursos

5 Desafíos de la Programación Concurrente en Producción

Desafío 1: Concurrencia Sin Límites Causa Agotamiento de Recursos

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

Cada URL genera una goroutine — 10,000 URLs significa 10,000 conexiones concurrentes. Los pools de conexiones a base de datos se agotan, los servicios descendentes se colapsan, la memoria se dispara.

Desafío 2: Fugas de Goroutines

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

Si el channel nunca se cierra, esta goroutine nunca sale. En servicios de larga duración, las goroutines filtradas se acumulan continuamente.

Desafío 3: Errores Silenciosamente Ignorados

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()
}

Los errores solo se registran — el llamador no tiene idea de los fallos. Si 2 de 3 URLs fallan, el llamador asume que todas tuvieron éxito.

Desafío 4: Sin Apagado Graceful

Cuando el servicio recibe SIGTERM, las goroutines en ejecución son interrumpidas forzosamente. Las escrituras en curso pueden corromper datos, las transacciones en progreso pueden quedar incompletas.

Desafío 5: Dificultad para Combinar Patrones de Concurrencia

Worker Pool necesita limitación de tasa, Pipeline necesita propagación de errores, Fan-out necesita agregación de resultados. Implementar cada patrón por separado no es difícil, pero combinarlos en un solo servicio crea interacciones que fácilmente llevan a deadlocks.

7 Patrones de Concurrencia de Grado de Producción

Patrón 1: Worker Pool — Pool de Goroutines Limitado

Worker Pool es el patrón de concurrencia más fundamental e importante. Idea central: un número fijo de workers extrae tareas de una cola de tareas, evitando concurrencia sin límites.

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()
}

Ejemplo de uso:

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()
}

Puntos clave de diseño:

  • Cantidad fija de workers previene explosión de goroutines
  • Channel de tareas con búfer sirve como cola de tareas
  • Soporte de Context para cancelación, los workers pueden salir de forma graceful
  • Submit es no bloqueante, retorna false cuando la cola está llena

Patrón 2: Fan-out/Fan-in — Dispersión-Recolección Paralela

Fan-out distribuye una fuente de datos a múltiples goroutines para procesamiento paralelo, Fan-in fusiona resultados de múltiples goroutines en un solo channel.

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
}

Ejemplo de uso — procesamiento paralelo de pedidos:

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

Puntos clave de diseño:

  • Soporte de genéricos (Go 1.18+), aplicable a cualquier tipo
  • Cada goroutine de fan-out consume independientemente del channel fuente
  • Fan-in usa WaitGroup para esperar que todos los canales de entrada se cierren
  • Todas las goroutines pueden salir con la cancelación del context

Patrón 3: Pipeline — Procesamiento por Etapas

Pipeline descompone un procesamiento complejo en etapas, cada una ejecutándose como una goroutine conectada por channels.

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
    }
}

Ejemplo de uso — pipeline de procesamiento de datos:

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)
        }
    }()
}

Puntos clave de diseño:

  • Cada etapa es una goroutine independiente, puede escalarse independientemente
  • Los channels proporcionan backpressure entre etapas
  • Los errores se manejan dentro de las etapas, no interrumpen todo el pipeline
  • Todas las etapas salen de forma graceful con la cancelación del context

Patrón 4: errgroup — Manejo Concurrente de Errores

errgroup es la versión con manejo de errores de sync.WaitGroup: el primer error cancela todas las 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 con límite de concurrencia:

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
}

Puntos clave de diseño:

  • errgroup.WithContext propaga automáticamente la cancelación
  • g.SetLimit(n) controla la concurrencia máxima (Go 1.20+)
  • El primer error cancela todas las goroutines en progreso
  • La captura de variables en closures requiere i, url := i, url

Patrón 5: Semáforo — Limitación de Tasa

semaphore.Weighted proporciona semáforos ponderados para asignación de recursos con diferentes pesos.

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()
}

Ejemplo de uso — limitación de tasa de API por niveles:

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()
    })
}

Semáforo con 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()
}

Puntos clave de diseño:

  • Semáforo ponderado, diferentes operaciones consumen diferentes cuotas
  • Acquire soporta cancelación de context y timeout
  • Release debe emparejarse con Acquire
  • Adecuado para limitación de tasa de API por niveles, control de cuota de recursos

Patrón 6: Cancelación de Context y Timeout

Context es la "línea de vida" de la programación concurrente en Go, utilizado para propagar señales de cancelación, timeouts y deadlines.

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
}

Patrón de apagado graceful:

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()
}

Puntos clave de diseño:

  • WithTimeout/WithDeadline para establecer timeouts
  • WithCancel para control manual de cancelación
  • signal.NotifyContext para escuchar señales del sistema
  • La cancelación del context se propaga a todas las goroutines hijas
  • defer cancel() previene fugas de context

Patrón 7: Servicio Concurrente en Producción — Combinando Patrones

Combina los patrones anteriores para construir un servicio concurrente de grado de producción: 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
}

Puntos clave de diseño:

  • errgroup controla concurrencia + propagación de errores
  • Channels para distribución de tareas y recolección de resultados
  • Context para timeout y cancelación unificados
  • WaitGroup asegura que todos los workers completen antes de cerrar el channel de resultados
  • Diseño por capas: planificación (errgroup) + ejecución (goroutines worker) + recolección (range resultCh)

5 Errores Comunes y Sus Soluciones

Error 1: Captura de Variables en Closures

❌ Incorrecto:

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

Todas las goroutines comparten la misma variable url, terminando por obtener la última URL.

✅ Correcto:

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

Error 2: Channel Sin Cerrar Causa Fuga de Goroutine

❌ Incorrecto:

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

Los consumidores que iteran sobre el channel nunca terminan, fuga de goroutine.

✅ Correcto:

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

Error 3: Posición Incorrecta de WaitGroup Add

❌ Incorrecto:

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

Las goroutines pueden no haber alcanzado wg.Add(1) aún, y el wg.Wait() de la goroutine principal pasa inmediatamente.

✅ Correcto:

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

Error 4: Fuga de time.After en Select

❌ Incorrecto:

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

Cada select crea un nuevo channel time.After. Llamadas frecuentes filtran timers.

✅ Correcto:

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

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

Error 5: Copia de Mutex

❌ Incorrecto:

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

func copyMap(m SafeMap) SafeMap {
    return m
}

El valor cero de Mutex es "desbloqueado". Copiar un Mutex bloqueado causa deadlock o data race.

✅ Correcto:

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

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

Referencia de Solución de Errores

Síntoma Causa Posible Investigación Solución
fatal error: all goroutines are asleep - deadlock! Todas las goroutines bloqueadas, sin goroutines activas Verificar emparejamiento lectura/escritura de channels, select default Asegurar que los channels tengan productores y consumidores, agregar cancelación de context
El conteo de goroutines sigue creciendo Fuga de goroutines, channels sin cerrar Monitoreo con runtime.NumGoroutine(), pprof goroutine profile Asegurar que todas las goroutines tengan rutas de salida, defer close(ch)
La memoria sigue creciendo Goroutines mantienen referencias a objetos grandes, acumulación en channels pprof heap profile, verificar tamaño de búfer de channels Limitar búfer de channels, liberar referencias oportunamente
Errores frecuentes de context canceled Context ascendente cancelado, timeout demasiado corto Verificar cadena de contexts, confirmar que el timeout es razonable Ajustar timeout, distinguir errores de negocio de errores de timeout
Resultados concurrentes faltantes Goroutine sale antes de escribir al channel Verificar lógica de salida de goroutines, confirmar cierre de channels Usar WaitGroup para asegurar que todas las goroutines completen antes de cerrar el channel
race condition detectado Múltiples goroutines leen/escriben variables compartidas go test -race, verificar variables globales y capturas en closures Usar Mutex/RWMutex para protección, o cambiar a comunicación por channels
Channel bloqueado, sin respuesta Desajuste de velocidad entre productor/consumidor Verificar tamaño de búfer del channel, monitorear tasas de producción/consumo Aumentar búfer, usar select+default para envío no bloqueante
Fallo en apagado graceful Goroutines no responden a la cancelación del context Verificar si las goroutines hacen select en ctx.Done() Asegurar que todas las goroutines de larga duración verifiquen ctx.Done()
errgroup retorna solo un error errgroup diseñado para cancelación por primer error Verificar si necesitas recolectar todos los errores Usar colector de errores personalizado o múltiples errgroups
Timeout en adquisición de semáforo Límite de concurrencia demasiado estricto, solicitudes encoladas demasiado tiempo Monitorear tiempo de espera del semáforo, ajustar cuotas de peso Aumentar capacidad del semáforo, optimizar asignación de pesos

Técnicas Avanzadas de Optimización

Optimización 1: Worker Pool Dinámico

Ajustar dinámicamente la cantidad de workers según la carga del sistema:

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()
}

Optimización 2: Batching

Combinar múltiples tareas pequeñas en una operación por lotes para reducir overhead de syscalls e 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
}

Optimización 3: Paso por Channel Sin Copia (Zero-Copy)

Usar punteros y sync.Pool para reducir la asignación de memoria durante el paso por channels:

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
}

Comparación de Patrones de Concurrencia

Característica goroutine+channel sync.WaitGroup errgroup Worker Pool
Control de concurrencia Sin control integrado Sin control integrado SetLimit Cantidad fija de workers
Manejo de errores Implementación manual Implementación manual Auto cancelación por primer error Implementación manual
Backpressure Búfer de channel Ninguno Ninguno Channel de tareas
Propagación de cancelación Context manual Context manual Context automático Context manual
Recolección de resultados Recepción por channel Variables compartidas Retorna slice Recepción por channel
Caso de uso Procesamiento de datos en flujo Espera de tareas por lotes Llamadas API concurrentes Procesamiento de tareas con límite de tasa
Complejidad Media Baja Baja Media
Cantidad de goroutines Variable Variable Limitada Fija
Recomendación para producción ⭐⭐⭐ ⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐

Resumen

La programación concurrente en Go no se trata de "simplemente hacer que funcione" — se trata de responder cuatro preguntas: ¿Cuántas goroutines se están ejecutando? ¿Cuándo salen? ¿Qué pasa cuando hay un error? ¿Cómo se recuperan los recursos? Worker Pool responde "cuántas", Context responde "cuándo salir", errgroup responde "qué hacer en caso de error", y defer close() responde "cómo recuperar". Domina estos 7 patrones, y tendrás la metodología central para la programación concurrente en Go de grado de producción.

Herramientas Recomendadas

  • Formateador JSON — Formatea respuestas JSON de servicios concurrentes, depura rápidamente problemas de estructura de datos
  • Calculadora de Hash — Calcula firmas de peticiones y checksums de datos, asegura consistencia de datos en peticiones concurrentes
  • Codificación/Decodificación Base64 — Maneja codificación de datos binarios en servicios concurrentes

Prueba estas herramientas que se ejecutan en tu navegador — no requieren registro →

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