Go desde cero, parte 13: patrones de concurrencia (worker pools, fan-in, fan-out y pipelines)

> serie: aprende go de 0 a 100 | artículo 13 de 19

---

Ya tienes goroutines y channels. Hoy los combinamos en los tres patrones de concurrencia que resuelven el 90% de los problemas concurrentes reales. Son los mismos planos que usa la industria, documentados en el blog oficial de Go y en la charla clásica de Rob Pike Go Concurrency Patterns.

---

Pipeline: la cadena de montaje

Un pipeline es una serie de etapas conectadas por channels: cada etapa recibe de la anterior, transforma y envía a la siguiente (Go Concurrency Patterns: Pipelines):

// Etapa 1: genera números.
func genera(nums ...int) <-chan int {
  out := make(chan int)
  go func() {
    defer close(out) // el emisor cierra
    for _, n := range nums {
      out <- n
    }
  }()
  return out
}

// Etapa 2: eleva al cuadrado.
func cuadrado(in <-chan int) <-chan int {
  out := make(chan int)
  go func() {
    defer close(out)
    for n := range in {
      out <- n * n
    }
  }()
  return out
}

// Tubería completa: se lee como una frase.
for v := range cuadrado(genera(2, 3, 4)) {
  fmt.Println(v) // 4 9 16
}

Fíjate en el patrón de cada etapa: crea su channel de salida, lanza su goroutine, devuelve el channel. El defer close(out) respeta la regla de oro del artículo 12: cierra quien envía. Cada etapa corre en paralelo con las demás, como una cadena de montaje de verdad, mientras una pieza se pinta, la siguiente ya se está cortando.

Worker pool: N obreros, una cola

El patrón más rentable en el trabajo real: limitar cuántas tareas procesas en paralelo (llamadas a APIs, consultas, archivos) sin frenar la entrada:

func worker(id int, jobs <-chan int, results chan<- int) {
  for j := range jobs { // reparto automático
    results <- j * 2 // trabajo duro simulado
  }
}

jobs := make(chan int, 100)
results := make(chan int, 100)

// 3 workers compiten por los mismos jobs.
for w := 1; w <= 3; w++ {
  go worker(w, jobs, results)
}

for j := 1; j <= 9; j++ {
  jobs <- j
}
close(jobs) // no hay más trabajo

for i := 0; i < 9; i++ {
  fmt.Println(<-results)
}

La magia está en que varios workers hagan range sobre el MISMO channel: cada job lo coge exactamente uno, el reparto es automático y gratis. ¿El equivalente en Java? Un ExecutorService con su cola y su factoría de threads. Aquí son quince líneas de stdlib.

Fan-out y fan-in: repartir y juntar

El worker pool ya esconde el fan-out: varios lectores del mismo channel repartiéndose el trabajo. Su gemelo es el fan-in: fusionar varias fuentes en un solo channel (blog de pipelines):

// fan-in: fusionar varios channels en uno.
func fusiona(entradas ...<-chan int) <-chan int {
  out := make(chan int)
  var wg sync.WaitGroup
  wg.Add(len(entradas))
  for _, in := range entradas {
    go func(c <-chan int) {
      defer wg.Done()
      for v := range c {
        out <- v
      }
    }(in)
  }
  go func() {
    wg.Wait() // cuando todas acaban...
    close(out) // ...se cierra la salida
  }()
  return out
}

Aquí aparece sync.WaitGroup, adelantándose a su artículo (el 14): es un contador de goroutines vivas. El problema que resuelve es fino: la salida solo puede cerrarse cuando TODAS las entradas terminaron, y eso no lo sabe ninguna goroutine copiadora por sí sola.

Los tres, juntos

Un servicio real encadena los tres sin pestañear: genera URLs (pipeline), las reparte entre N workers que descargan (fan-out), fusiona los resultados (fan-in) y sigue procesando aguas abajo. Cada pieza es pequeña, testeable y sin locks a la vista. Estos son los patrones de concurrencia que cumplen la promesa CSP del artículo 11: la estructura del programa ES la estructura de la concurrencia.

Aviso de honestidad técnica, falta un ingrediente. ¿Qué pasa si el consumidor abandona a mitad y los workers quedan bloqueados enviando a un channel que nadie lee? Eso es una fuga de goroutines, y su solución oficial (cancelación con context) es el plato principal del próximo artículo. El blog de pipelines cubre la versión con channel done si no puedes esperar.

---

Reto de la semana: pon a prueba los patrones de concurrencia

• Monta un pipeline de tres etapas: genera 1-20, filtra pares, eleva al cubo. Cada etapa con su goroutine y su defer close.
• Worker pool de 4 workers que calcule si cada número de una lista de 50 es primo. Compara el tiempo con 1 worker frente a 4 (pista: time.Since).
• Implementa fusiona y únela a dos generadores independientes. Verifica con -race.
• Bonus: rompe algo a propósito. Quita el close(jobs) del worker pool y observa el deadlock que el runtime detecta y reporta. Conocer el error de cerca vacuna.

---

Patrones de concurrencia: nivel completado.

En el próximo artículo: sync, context y cancelación. Cómo decirle educadamente a cien goroutines que se ha acabado la fiesta.

> exit