Fundamentos Avanzados de Concurrencia

10 ejemplos para empezar con Concurrencia Avanzada - 7 básicos y 3 intermedios.

Prerrequisitos

Go 1.26.x o posterior y un módulo: go mod init example.com/advconc.
Instala extensiones: go get golang.org/x/sync/errgroup golang.org/x/time/rate.
Ejecuta los ejemplos con go run . desde el directorio del paquete de cada fragmento.
La familiaridad con goroutines y canales de Fundamentos de Concurrencia es útil.

Ejemplos Básicos

1. Pool de trabajadores fijo con canal de trabajos

Inicia N trabajadores que leen trabajos hasta que el canal se cierra.

package main
 
import (
    "fmt"
    "sync"
)
 
func worker(id int, jobs <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs {
        fmt.Printf("trabajador %d trabajo %d\n", id, j)
    }
}
 
func main() {
    const workers = 3
    jobs := make(chan int, 10)
    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go worker(i, jobs, &wg)
    }
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    close(jobs)
    wg.Wait()
}

close(jobs) señala a los trabajadores que salgan después de procesar todo.
El tamaño del buffer desacopla a los productores brevemente; un buffer cero sincroniza cada envío.
WaitGroup espera a los trabajadores antes de que main retorne.

Relacionado: Pools de Trabajadores y Colas de Tareas - patrones de dimensionamiento y apagado

2. Semáforo con canal con buffer

Limita las goroutines en ejecución sin un conjunto de trabajadores dedicado.

package main
 
import (
    "fmt"
    "sync"
)
 
func main() {
    sem := make(chan struct{}, 2) // máx 2 concurrentes
    var wg sync.WaitGroup
    for i := 0; i < 6; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            sem <- struct{}{}
            defer func() { <-sem }()
            fmt.Println("tarea", n)
        }(i)
    }
    wg.Wait()
}

Enviar adquiere una ranura; recibir la libera.
La capacidad es igual al paralelismo máximo - un semáforo de conteo idiomático.
Prefiere golang.org/x/sync/semaphore cuando necesites Acquire(ctx) con cancelación.

Relacionado: Limitación de Tasa y Backpressure - cubos de tokens y descarte

3. errgroup con tareas paralelas

Ejecuta goroutines relacionadas; el primer error cancela las demás.

package main
 
import (
    "context"
    "fmt"
    "time"
 
    "golang.org/x/sync/errgroup"
)
 
func main() {
    g, ctx := errgroup.WithContext(context.Background())
    g.Go(func() error {
        return fmt.Errorf("la tarea a falló")
    })
    g.Go(func() error {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(time.Second):
            return nil
        }
    })
    if err := g.Wait(); err != nil {
        fmt.Println("errgroup:", err)
    }
}

WithContext cancela ctx cuando cualquier Go devuelve un error no nulo.
Respeta siempre ctx en funciones Go de larga duración.
Wait bloquea hasta que todas las goroutines terminan y devuelve el primer error.

Relacionado: errgroup y Extensiones de golang.org/x/sync - SetLimit y patrones

4. Pipeline de dos etapas

Una goroutine produce, otra consume a través de un canal.

package main
 
import (
    "fmt"
)
 
func main() {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := 1; i <= 3; i++ {
            out <- i * 10
        }
    }()
    for v := range out {
        fmt.Println(v)
    }
}

El productor es el propietario del canal y debe cerrarlo cuando termina.
range sobre el canal recibe hasta que se cierra.
Añade buffer entre etapas para reducir el bloqueo cuando las tasas difieren.

Relacionado: Fan-In, Fan-Out y Etapas de Pipeline - grafos multi-etapa

5. Fan-out con WaitGroup

Distribuye elementos a los trabajadores, espera la finalización.

package main
 
import (
    "fmt"
    "sync"
)
 
func main() {
    items := []int{1, 2, 3, 4}
    var wg sync.WaitGroup
    for _, item := range items {
        wg.Add(1)
        go func(v int) {
            defer wg.Done()
            fmt.Println("procesar", v)
        }(item)
    }
    wg.Wait()
}

Captura la variable de bucle en el parámetro de cierre para evitar el error clásico de variable de bucle.
El fan-out ilimitado está bien para N pequeños; usa un pool para lotes grandes.
Los errores necesitan una estrategia de agregación separada (errgroup o canal de errores).

Relacionado: Concurrencia en Producción: Patrones Más Allá de Hello Goroutine - cuándo limitar el fan-out

6. Bucle de trabajador cancelado por contexto

Detiene a los trabajadores cuando el contexto se completa.

package main
 
import (
    "context"
    "fmt"
    "time"
)
 
func worker(ctx context.Context, id int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("trabajador", id, "detener")
            return
        default:
            fmt.Println("trabajador", id, "tick")
            time.Sleep(200 * time.Millisecond)
        }
    }
}
 
func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
    defer cancel()
    go worker(ctx, 1)
    <-ctx.Done()
    time.Sleep(100 * time.Millisecond)
}

select sobre ctx.Done() es obligatorio en bucles de producción.
WithTimeout deriva una fecha límite del contexto padre.
Empareja con WaitGroup cuando múltiples trabajadores deban salir antes de continuar.

Relacionado: Fundamentos de context - WithCancel y fechas límite

7. Incremento de contador atómico

Contador sin bloqueos para métricas e IDs.

package main
 
import (
    "fmt"
    "sync"
    "sync/atomic"
)
 
func main() {
    var n atomic.Int64
    var wg sync.WaitGroup
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            n.Add(1)
        }()
    }
    wg.Wait()
    fmt.Println("conteo", n.Load())
}

atomic.Int64 está tipado y es más claro que int64 crudo con AddInt64.
Usa atómicos para contadores simples; usa mutex para structs con múltiples campos.
Ejecuta go test -race si mezclas atómicos con lecturas no atómicas.

Relacionado: Operaciones sync/atomic - CAS y cuándo ganan los atómicos

Ejemplos Intermedios

8. Pool de trabajadores con apagado por contexto

Parada elegante: cancela el contexto, cierra los trabajos, espera a los trabajadores.

package main
 
import (
    "context"
    "fmt"
    "sync"
    "time"
)
 
func pool(ctx context.Context, jobs <-chan int, n int) *sync.WaitGroup {
    var wg sync.WaitGroup
    for i := 0; i < n; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case j, ok := <-jobs:
                    if !ok {
                        return
                    }
                    fmt.Println(id, "manejado", j)
                }
            }
        }(i)
    }
    return &wg
}
 
func main() {
    ctx, cancel := context.WithCancel(context.Background())
    jobs := make(chan int, 4)
    wg := pool(ctx, jobs, 2)
    for j := 1; j <= 4; j++ {
        jobs <- j
    }
    close(jobs)
    cancel()
    wg.Wait()
}

La combinación de ctx.Done() y !ok en los trabajos cubre tanto las rutas de cancelación como de vaciado.
Cierra el canal de trabajos solo después de que los productores terminen de enviar.
wg.Wait() antes de salir confirma que los trabajadores se detuvieron.

Relacionado: Evitar Fugas de Goroutines y Deadlocks de Canales - reglas de propiedad

9. errgroup para fetches paralelos estilo HTTP

Simula I/O paralelo con cancelación compartida ante el primer fallo.

package main
 
import (
    "context"
    "errors"
    "fmt"
 
    "golang.org/x/sync/errgroup"
)
 
func fetch(ctx context.Context, name string, fail bool) error {
    if fail {
        return fmt.Errorf("%s no disponible", name)
    }
    select {
    case <-ctx.Done():
        return ctx.Err()
    default:
        fmt.Println("ok", name)
        return nil
    }
}
 
func main() {
    g, ctx := errgroup.WithContext(context.Background())
    g.Go(func() error { return fetch(ctx, "usuarios", false) })
    g.Go(func() error { return fetch(ctx, "facturacion", true) })
    if err := g.Wait(); err != nil {
        fmt.Println("abortar:", err)
    }
}

El primer fetch fallido cancela ctx para los hermanos que aún se ejecutan.
Mapea este patrón a http.NewRequestWithContext en servicios reales.
Usa g.SetLimit(n) cuando el fan-out deba respetar las cuotas del downstream.

Relacionado: Propagación de Cancelación en Manejadores HTTP - ctx con ámbito de solicitud

10. Productor con tasa limitada

El cubo de tokens limita la velocidad a la que los trabajos entran en el pool.

package main
 
import (
    "context"
    "fmt"
    "time"
 
    "golang.org/x/time/rate"
)
 
func main() {
    lim := rate.NewLimiter(rate.Every(100*time.Millisecond), 1)
    ctx := context.Background()
    for i := 1; i <= 5; i++ {
        if err := lim.Wait(ctx); err != nil {
            panic(err)
        }
        fmt.Println("emitir", i, time.Now().Format("15:04:05.000"))
    }
}

Every(d) establece la tasa constante; burst permite picos cortos.
Wait bloquea; Allow no bloquea para rutas de carga descartada.
Pasa el ctx de la solicitud a Wait para que la cancelación detenga a un productor limitado.

Relacionado: Limitación de Tasa y Backpressure - semáforos vs cubo de tokens

Versiones de la Pila: Esta página fue escrita para Go 1.26.x (GC por defecto Green Tea, go fix modernizers - verifica el parche en la compilación), chi (última - verifica en la compilación), gin (última - verifica en la compilación), echo (última - verifica en la compilación), google.golang.org/grpc (última - verifica en la compilación), sigs.k8s.io/controller-runtime (última - verifica en la compilación), kubebuilder (última - verifica en la compilación), tinygo (última - verifica los objetivos de la placa en la compilación), wazero (última - verifica en la compilación), y golangci-lint (última - verifica el conjunto de linters en la compilación).

Prerrequisitos

Go 1.26.x o posterior y un módulo: go mod init example.com/advconc.
Instala extensiones: go get golang.org/x/sync/errgroup golang.org/x/time/rate.
Ejecuta los ejemplos con go run . desde el directorio del paquete de cada fragmento.
La familiaridad con goroutines y canales de Fundamentos de Concurrencia es útil.

Ejemplos Básicos

1. Pool de trabajadores fijo con canal de trabajos

Inicia N trabajadores que leen trabajos hasta que el canal se cierra.

package main
 
import (
    "fmt"
    "sync"
)
 
func worker(id int, jobs <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs {
        fmt.Printf("trabajador %d trabajo %d\n", id, j)
    }
}
 
func main() {
    const workers = 3
    jobs := make(chan int, 10)
    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go worker(i, jobs, &wg)
    }
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    close(jobs)
    wg.Wait()
}

close(jobs) señala a los trabajadores que salgan después de procesar todo.
El tamaño del buffer desacopla a los productores brevemente; un buffer cero sincroniza cada envío.
WaitGroup espera a los trabajadores antes de que main retorne.

Relacionado: Pools de Trabajadores y Colas de Tareas - patrones de dimensionamiento y apagado

2. Semáforo con canal con buffer

Limita las goroutines en ejecución sin un conjunto de trabajadores dedicado.

package main
 
import (
    "fmt"
    "sync"
)
 
func main() {
    sem := make(chan struct{}, 2) // máx 2 concurrentes
    var wg sync.WaitGroup
    for i := 0; i < 6; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            sem <- struct{}{}
            defer func() { <-sem }()
            fmt.Println("tarea", n)
        }(i)
    }
    wg.Wait()
}

Enviar adquiere una ranura; recibir la libera.
La capacidad es igual al paralelismo máximo - un semáforo de conteo idiomático.
Prefiere golang.org/x/sync/semaphore cuando necesites Acquire(ctx) con cancelación.

Relacionado: Limitación de Tasa y Backpressure - cubos de tokens y descarte

3. errgroup con tareas paralelas

Ejecuta goroutines relacionadas; el primer error cancela las demás.

package main
 
import (
    "context"
    "fmt"
    "time"
 
    "golang.org/x/sync/errgroup"
)
 
func main() {
    g, ctx := errgroup.WithContext(context.Background())
    g.Go(func() error {
        return fmt.Errorf("la tarea a falló")
    })
    g.Go(func() error {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(time.Second):
            return nil
        }
    })
    if err := g.Wait(); err != nil {
        fmt.Println("errgroup:", err)
    }
}

WithContext cancela ctx cuando cualquier Go devuelve un error no nulo.
Respeta siempre ctx en funciones Go de larga duración.
Wait bloquea hasta que todas las goroutines terminan y devuelve el primer error.

Relacionado: errgroup y Extensiones de golang.org/x/sync - SetLimit y patrones

4. Pipeline de dos etapas

Una goroutine produce, otra consume a través de un canal.

package main
 
import (
    "fmt"
)
 
func main() {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := 1; i <= 3; i++ {
            out <- i * 10
        }
    }()
    for v := range out {
        fmt.Println(v)
    }
}

El productor es el propietario del canal y debe cerrarlo cuando termina.
range sobre el canal recibe hasta que se cierra.
Añade buffer entre etapas para reducir el bloqueo cuando las tasas difieren.

Relacionado: Fan-In, Fan-Out y Etapas de Pipeline - grafos multi-etapa

5. Fan-out con WaitGroup

Distribuye elementos a los trabajadores, espera la finalización.

package main
 
import (
    "fmt"
    "sync"
)
 
func main() {
    items := []int{1, 2, 3, 4}
    var wg sync.WaitGroup
    for _, item := range items {
        wg.Add(1)
        go func(v int) {
            defer wg.Done()
            fmt.Println("procesar", v)
        }(item)
    }
    wg.Wait()
}

Captura la variable de bucle en el parámetro de cierre para evitar el error clásico de variable de bucle.
El fan-out ilimitado está bien para N pequeños; usa un pool para lotes grandes.
Los errores necesitan una estrategia de agregación separada (errgroup o canal de errores).

Relacionado: Concurrencia en Producción: Patrones Más Allá de Hello Goroutine - cuándo limitar el fan-out

6. Bucle de trabajador cancelado por contexto

Detiene a los trabajadores cuando el contexto se completa.

package main
 
import (
    "context"
    "fmt"
    "time"
)
 
func worker(ctx context.Context, id int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("trabajador", id, "detener")
            return
        default:
            fmt.Println("trabajador", id, "tick")
            time.Sleep(200 * time.Millisecond)
        }
    }
}
 
func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
    defer cancel()
    go worker(ctx, 1)
    <-ctx.Done()
    time.Sleep(100 * time.Millisecond)
}

select sobre ctx.Done() es obligatorio en bucles de producción.
WithTimeout deriva una fecha límite del contexto padre.
Empareja con WaitGroup cuando múltiples trabajadores deban salir antes de continuar.

Relacionado: Fundamentos de context - WithCancel y fechas límite

7. Incremento de contador atómico

Contador sin bloqueos para métricas e IDs.

package main
 
import (
    "fmt"
    "sync"
    "sync/atomic"
)
 
func main() {
    var n atomic.Int64
    var wg sync.WaitGroup
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            n.Add(1)
        }()
    }
    wg.Wait()
    fmt.Println("conteo", n.Load())
}

atomic.Int64 está tipado y es más claro que int64 crudo con AddInt64.
Usa atómicos para contadores simples; usa mutex para structs con múltiples campos.
Ejecuta go test -race si mezclas atómicos con lecturas no atómicas.

Relacionado: Operaciones sync/atomic - CAS y cuándo ganan los atómicos

Ejemplos Intermedios

8. Pool de trabajadores con apagado por contexto

Parada elegante: cancela el contexto, cierra los trabajos, espera a los trabajadores.

package main
 
import (
    "context"
    "fmt"
    "sync"
    "time"
)
 
func pool(ctx context.Context, jobs <-chan int, n int) *sync.WaitGroup {
    var wg sync.WaitGroup
    for i := 0; i < n; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case j, ok := <-jobs:
                    if !ok {
                        return
                    }
                    fmt.Println(id, "manejado", j)
                }
            }
        }(i)
    }
    return &wg
}
 
func main() {
    ctx, cancel := context.WithCancel(context.Background())
    jobs := make(chan int, 4)
    wg := pool(ctx, jobs, 2)
    for j := 1; j <= 4; j++ {
        jobs <- j
    }
    close(jobs)
    cancel()
    wg.Wait()
}

La combinación de ctx.Done() y !ok en los trabajos cubre tanto las rutas de cancelación como de vaciado.
Cierra el canal de trabajos solo después de que los productores terminen de enviar.
wg.Wait() antes de salir confirma que los trabajadores se detuvieron.

Relacionado: Evitar Fugas de Goroutines y Deadlocks de Canales - reglas de propiedad

9. errgroup para fetches paralelos estilo HTTP

Simula I/O paralelo con cancelación compartida ante el primer fallo.

package main
 
import (
    "context"
    "errors"
    "fmt"
 
    "golang.org/x/sync/errgroup"
)
 
func fetch(ctx context.Context, name string, fail bool) error {
    if fail {
        return fmt.Errorf("%s no disponible", name)
    }
    select {
    case <-ctx.Done():
        return ctx.Err()
    default:
        fmt.Println("ok", name)
        return nil
    }
}
 
func main() {
    g, ctx := errgroup.WithContext(context.Background())
    g.Go(func() error { return fetch(ctx, "usuarios", false) })
    g.Go(func() error { return fetch(ctx, "facturacion", true) })
    if err := g.Wait(); err != nil {
        fmt.Println("abortar:", err)
    }
}

El primer fetch fallido cancela ctx para los hermanos que aún se ejecutan.
Mapea este patrón a http.NewRequestWithContext en servicios reales.
Usa g.SetLimit(n) cuando el fan-out deba respetar las cuotas del downstream.

Relacionado: Propagación de Cancelación en Manejadores HTTP - ctx con ámbito de solicitud

10. Productor con tasa limitada

El cubo de tokens limita la velocidad a la que los trabajos entran en el pool.

package main
 
import (
    "context"
    "fmt"
    "time"
 
    "golang.org/x/time/rate"
)
 
func main() {
    lim := rate.NewLimiter(rate.Every(100*time.Millisecond), 1)
    ctx := context.Background()
    for i := 1; i <= 5; i++ {
        if err := lim.Wait(ctx); err != nil {
            panic(err)
        }
        fmt.Println("emitir", i, time.Now().Format("15:04:05.000"))
    }
}

Every(d) establece la tasa constante; burst permite picos cortos.
Wait bloquea; Allow no bloquea para rutas de carga descartada.
Pasa el ctx de la solicitud a Wait para que la cancelación detenga a un productor limitado.

Relacionado: Limitación de Tasa y Backpressure - semáforos vs cubo de tokens

Versiones de la Pila: Esta página fue escrita para Go 1.26.x (GC por defecto Green Tea, go fix modernizers - verifica el parche en la compilación), chi (última - verifica en la compilación), gin (última - verifica en la compilación), echo (última - verifica en la compilación), google.golang.org/grpc (última - verifica en la compilación), sigs.k8s.io/controller-runtime (última - verifica en la compilación), kubebuilder (última - verifica en la compilación), tinygo (última - verifica los objetivos de la placa en la compilación), wazero (última - verifica en la compilación), y golangci-lint (última - verifica el conjunto de linters en la compilación).

Fundamentos Avanzados de Concurrencia

Buscar en la documentación

Fundamentos Avanzados de Concurrencia