Pool de workers en Go: límite de goroutines, resultados y cancelación segura

1. Introducción

Las goroutines en Go son muy baratas, así que llamar a go para cada tarea es común. Cuando el número de tareas llega a decenas de miles y cada tarea llama a la base de datos o HTTP, este enfoque fácilmente agota el pool de conexiones o excede el límite de velocidad del servicio externo.

Este artículo te guía en la construcción de un pool de workers, es decir, un número fijo de goroutines que reciben trabajo de un canal. Después de leerlo, sabrás cómo escribir un pool básico, recopilar resultados y errores, cancelar con context, simplificar con errgroup.SetLimit, y verificar que el pool realmente limita el nivel de paralelismo.

2. El problema: goroutines sin límite

Considera el siguiente código que procesa una lista de pedidos:

for _, o := range orders {
	go chargeOrder(o) // 50.000 pedidos -> 50.000 goroutines llamando simultáneamente a la API de pago
}

Cada goroutine consume solo unos pocos KB de memoria, por lo que el problema generalmente no está en la RAM. El problema está en los recursos del backend:

• El pool de conexión de la base de datos tiene alrededor de 20–50 conexiones.
• El gateway de pago limita el número de solicitudes por segundo.
• El número de descriptores de archivo del proceso tiene límite.

El pool de workers resuelve esto separando dos conceptos. El número de tareas puede ser muy grande, pero el número de tareas ejecutándose simultáneamente está limitado por el número de workers.

3. Pool de workers básico

El modelo consta de tres partes:

1. Un channel jobs que contiene los trabajos a realizar.
2. N goroutines worker, cada worker itera con for range sobre el channel.
3. Un sync.WaitGroup para esperar a que todos los workers terminen.

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) // simular llamada a API de pago
		fmt.Printf("worker %d procesando order %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("completado en", time.Since(start).Round(100*time.Millisecond))
}

El código funciona de la siguiente manera:

• orders <-chan Order es un channel de solo recepción desde la perspectiva del worker. El compilador bloqueará al worker si intenta enviar o cerrar el channel.
• for o := range orders se cierra automáticamente cuando el channel se close y se han agotado todos los elementos.
• close(orders) es la señal de “fin del trabajo”. Si olvidas esta línea, el worker espera para siempre y wg.Wait() causa un deadlock.
• wg.Add(1) debe llamarse antes de go worker(...), no dentro de la goroutine.

Resultado de la ejecución (el orden de las líneas puede variar):

worker 3 procesando order 1 (10)
worker 1 procesando order 2 (20)
worker 2 procesando order 3 (30)
worker 2 procesando order 4 (40)
...
completado en 300ms

9 pedidos divididos entre 3 workers, cada pedido tarda 100ms, por lo que el tiempo total es alrededor de 300ms en lugar de 900ms si se ejecuta secuencialmente.

Nota: A partir de Go 1.25, puedes escribir wg.Go(func() { worker(i, orders) }). Esta función automáticamente llama Add y Done, evitando el error de olvidar uno de los dos.

4. Recopilación de resultados y errores

En la práctica, generalmente necesitas saber qué pedidos tuvieron éxito y cuáles fallaron. Agregamos un channel results para esto:

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("cantidad inválida")
	}
	return o.Amount * 0.02, nil // comisión 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("order %d error: %v\n", r.OrderID, r.Err)
			continue
		}
		fmt.Printf("order %d comisión %.2f\n", r.OrderID, r.Fee)
	}
}

La salida se ve así:

order 1 comisión 2.00
order 3 comisión 5.00
order 2 error: cantidad inválida
order 4 comisión 1.60

Hay tres puntos a tener en cuenta:

• El envío de trabajos se ejecuta en una goroutine separada. Si main envía trabajos y lee resultados en la misma goroutine, el worker se bloqueará en results <- mientras que main se bloquea en jobs <-. El resultado es un deadlock.
• La tercera goroutine espera wg.Wait() y luego close(results). Solo cuando todos los workers terminan podemos estar seguros de que nadie más está enviando a results. Regla general: solo el lado que envía cierra el channel.
• Los errores se incluyen en Result en lugar de detener todo el pool. El error de un pedido no debe bloquear otros pedidos.

El orden de los resultados no coincide necesariamente con el orden de entrada. Si necesitas preservar el orden, lleva el índice en el trabajo o usa el enfoque en la sección 6.

5. Cancelación de pool con context

Cuando una solicitud HTTP se cancela o agota el tiempo de espera, el pool debe detenerse en lugar de continuar llamando a la API de pago. Haces esto permitiendo que el worker escuche a 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
			}
		}
	}
}

El lado del productor también necesita un select similar:

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

Y crea un context con timeout donde lo llames:

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

La razón por la que necesitas select en ambas direcciones de envío: si el worker solo verifica ctx cuando recibe un trabajo, puede quedarse atrapado en results <- cuando el lado de lectura ya se ha ido. Esta es una fuente muy común de goroutine leak.

Nota: select solo ayuda al worker a salir entre trabajos. Si chargeOrder hace una llamada HTTP, necesitas pasar ctx a ella (http.NewRequestWithContext) para poder cancelar la solicitud que está en curso.

6. Simplificación con errgroup.SetLimit

Escribir manualmente channels, WaitGroup y context es propenso a errores. El paquete errgroup tiene SetLimit incorporado, que permite crear un pool limitado en solo unas pocas líneas. Primero, añade la dependencia:

go get golang.org/x/sync/errgroup

La siguiente función genérica procesa un slice con un máximo de limit goroutines, preserva el orden de los resultados y se detiene en el primer error:

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
}

El código funciona de la siguiente manera:

• g.SetLimit(limit) hace que g.Go se bloquee cuando hay suficientes goroutines limit en ejecución. En términos de eficiencia, es equivalente a un worker pool.
• errgroup.WithContext devuelve un ctx que se cancelará tan pronto como una función devuelva un error. Las otras tareas que usan ctx recibirán una señal de detención.
• results[i] = r es seguro cuando se ejecuta concurrentemente porque cada goroutine escribe en un índice único. No se necesita mutex en este caso.
• A partir de Go 1.22, las variables de bucle i e item se crean nuevas en cada iteración, por lo que el closure ya no comparte variables como en versiones anteriores.

Forma de llamar:

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

La diferencia con un pool escrito manualmente: errgroup crea una nueva goroutine para cada elemento en lugar de reutilizar N workers. Para la mayoría de aplicaciones, esta diferencia es insignificante. Si necesitas que los workers mantengan un estado privado, como cada worker tener una conexión, entonces usa el pool escrito manualmente en las secciones 3–5.

7. Pruebas del pool de workers

Necesitas probar dos comportamientos principales. Primero, el pool nunca excede el limit. Segundo, los errores se devuelven correctamente. Para medir el nivel de concurrencia, usamos 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("error inesperado: %v", err)
	}
	if maxSeen.Load() > 3 {
		t.Errorf("ejecutando %d concurrentemente, límite es 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("tarjeta rechazada")

	_, 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("quería errDeclined, recibí %v", err)
	}
}

Ejecuta las pruebas con el race detector:

go test -race ./...

ok  	example.com/shop/pool	0.061s

El bucle CompareAndSwap actualiza maxSeen sin necesidad de un bloqueo. La bandera -race es obligatoria con código concurrente, ya que detecta accesos a datos compartidos que los ojos suelen pasar por alto.

8. Errores comunes y cuándo no usarlo

Errores comunes:

• Olvidar close(jobs): el worker nunca sale y el programa reporta fatal error: all goroutines are asleep - deadlock!.
• Cerrar el channel en el lado receptor, o cerrarlo dos veces: causa panic: close of closed channel.
• Llamar wg.Add dentro de la goroutine: Wait puede devolver antes de que la goroutine tenga oportunidad de Add.
• Enviar trabajos y leer resultados en la misma goroutine con un channel sin buffer: causa deadlock como se analizó en la sección 4.
• No hacer select en ctx.Done() cuando se envían resultados: conduce a goroutine leak cuando el lado de lectura ya se ha ido.
• Elegir el número de workers arbitrariamente: para tareas vinculadas a CPU, usa runtime.NumCPU(). Para tareas vinculadas a I/O, basa en el límite de los recursos del backend, por ejemplo, el tamaño del pool de conexión.

Cuándo no usar worker pool:

• Un pequeño número de tareas conocidas de antemano, por ejemplo llamar a 3 servicios en paralelo. En este caso, errgroup sin SetLimit es suficiente.
• Tareas muy breves, solo unos pocos microsegundos. El costo de pasar por el channel puede ser mayor que el trabajo en sí, por lo que el procesamiento secuencial o por lotes será más eficiente.
• Necesitas limitar por velocidad (requests/segundo) en lugar de por cantidad concurrente. Este caso necesita un rate limiter, como golang.org/x/time/rate. El worker pool no puede reemplazar a un rate limiter.

9. Conclusión

El worker pool te ayuda a controlar el nivel de concurrencia en Go: el número de tareas puede ser muy grande pero los recursos del backend siguen siendo protegidos. El artículo ha cubierto los siguientes pasos:

• Pool básico con channel y sync.WaitGroup.
• Recopilación de resultados y errores a través del channel results, cierre del channel en el lugar correcto.
• Cancelación segura con context en ambas direcciones de envío.
• Simplificación con errgroup.SetLimit usando la función Process genérica.
• Pruebas de los límites de concurrencia y errores con go test -race.

Para la mayoría de los casos, errgroup.SetLimit es la opción más simple y menos propensa a errores. Solo debes escribir el pool manualmente cuando los workers necesiten mantener un estado privado o cuando necesites control detallado del flujo de trabajos y resultados.