Patrones de concurrencia en APIs: worker pools y rate limiting

Introducción: La concurrencia como pilar del rendimiento

En el desarrollo de APIs de alto rendimiento con Go, la concurrencia no es una característica opcional, sino el núcleo mismo de su arquitectura. Mientras que otros lenguajes luchan con hilos del sistema operativo y complejos mecanismos de sincronización, Go ofrece goroutines y canales como primitivas de primera clase, convirtiendo la programación concurrente en un modelo accesible y poderoso. Sin embargo, esta facilidad inicial puede llevar a un error crítico: asumir que lanzar miles de goroutines para manejar cada petición HTTP es la estrategia óptima. La realidad es más matizada. Sin patrones de diseño adecuados, como los Worker Pools y el Rate Limiting, un sistema puede colapsar bajo su propia carga, agotando recursos, saturando bases de datos o siendo bloqueado por servicios externos.

Esta lección se centra en la aplicación práctica de estos dos patrones fundamentales dentro del contexto de un microservicio REST construido con gorilla/mux. No se trata solo de entender la sintaxis, sino de internalizar cuándo y por qué aplicar cada patrón para construir sistemas que sean no solo rápidos, sino también resilientes, predecibles y justos en el consumo de recursos. Dominar estos patrones es lo que separa a un desarrollador que escribe código que funciona de uno que diseña sistemas que escalan.

Concepto Clave: Entendiendo los Patrones desde sus Fundamentos

Imagina una cocina de un restaurante muy concurrido. Si cada nuevo pedido (petición HTTP) exigiera contratar a un nuevo chef (goroutine) desde cero, el caos sería absoluto: costes disparados, despensa agotada y colapso en la zona de cocción. Un Worker Pool es el equivalente a tener un equipo estable y finito de chefs expertos (los workers). Los pedidos llegan y se colocan en una barra (el job channel). Cada chef, cuando termina su plato anterior, toma el siguiente pedido de la barra. Este patrón limita el consumo máximo simultáneo de recursos (CPU, memoria, conexiones a BD), previniendo el agotamiento y permitiendo un procesamiento ordenado y eficiente.

Por otro lado, el Rate Limiting actúa como el maître del restaurante. Incluso con una cocina eficiente, no puedes dejar entrar a 100 comensales a la vez; el servicio se degradaría para todos. El maître controla el flujo de entrada, admitiendo un número máximo de comensales por minuto o haciendo esperar a los nuevos si el restaurante está lleno. En una API, este patrón limita la cantidad de peticiones que un cliente, un endpoint o todo el servicio puede procesar en una ventana de tiempo. Protege los recursos backend, mitiga ataques de fuerza bruta y asegura una calidad de servicio equitativa para todos los consumidores.

Tip Clave: Piensa en el Worker Pool como un control de concurrencia (cuántas tareas se ejecutan AL MISMO TIEMPO) y en el Rate Limiting como un control de rendimiento (cuántas tareas se ejecutan EN UN PERIODO DADO). Son complementarios y a menudo se usan juntos.

Cómo funciona en la práctica: Integración en una API con gorilla/mux

La integración de estos patrones en un router como gorilla/mux es elegante y no intrusiva. Para un Worker Pool, típicamente se inicializa al arrancar la aplicación. Este pool vive durante todo el ciclo de vida del servicio, escuchando un canal de trabajos. Cuando llega una petición HTTP a un endpoint que requiere procesamiento intensivo (por ejemplo, procesamiento de imágenes, envío masivo de emails, cálculos complejos), el handler no ejecuta la tarea de inmediato. En su lugar, construye una estructura de trabajo (un Job) que contiene los datos de la petición y un canal para devolver el resultado. Esta "tarea" se envía al canal del pool. Una goroutine worker disponible la recogerá, la procesará y enviará el resultado de vuelta a través del canal de respuesta, que el handler HTTP original estará esperando.

El Rate Limiting se implementa a menudo como un middleware. Un middleware en gorilla/mux es una función que se ejecuta antes (o después) del handler final de una ruta. Un middleware de rate limiting intercepta la petición, consulta un "registro" (que puede estar en memoria, en Redis, etc.) para contar cuántas peticiones ha hecho esta IP o este token de API en los últimos segundos. Si el contador está por debajo del límite, incrementa el contador y pasa el control al siguiente handler. Si se excede el límite, responde inmediatamente con un código HTTP 429 Too Many Requests y un header Retry-After, sin tocar los recursos de la aplicación. Este patrón puede aplicarse globalmente, por ruta o por cliente.

Código en acción: Implementación de un Worker Pool para Procesamiento Asíncrono

El siguiente ejemplo muestra un Worker Pool genérico que puede ser integrado en cualquier servicio. Los trabajos son funciones que se ejecutarán de forma concurrente pero controlada.

// worker_pool.go
package main

import (
    "fmt"
    "log"
    "net/http"
    "sync"
    "time"

    "github.com/gorilla/mux"
)

// Job representa una unidad de trabajo.
type Job struct {
    ID       int
    Task     func() (string, error) // La tarea a ejecutar.
    Result   chan<- string          // Canal para enviar el resultado.
    Error    chan<- error           // Canal para enviar errores.
}

// WorkerPool gestiona un conjunto de workers.
type WorkerPool struct {
    jobQueue   chan Job
    wg         sync.WaitGroup
    maxWorkers int
}

// NewWorkerPool crea un nuevo pool.
func NewWorkerPool(maxWorkers, queueSize int) *WorkerPool {
    pool := &WorkerPool{
        jobQueue:   make(chan Job, queueSize),
        maxWorkers: maxWorkers,
    }
    pool.wg.Add(maxWorkers)
    for i := 1; i <= maxWorkers; i++ {
        go pool.worker(i)
    }
    return pool
}

// worker es la función que ejecuta cada goroutine del pool.
func (wp *WorkerPool) worker(id int) {
    defer wp.wg.Done()
    for job := range wp.jobQueue {
        log.Printf("Worker %d procesando job %d", id, job.ID)
        result, err := job.Task()
        if err != nil {
            job.Error <- err
        } else {
            job.Result <- result
        }
    }
}

// Submit envía un trabajo al pool.
func (wp *WorkerPool) Submit(task func() (string, error)) (string, error) {
    resultChan := make(chan string, 1)
    errorChan := make(chan error, 1)

    job := Job{
        ID:     time.Now().Nanosecond(),
        Task:   task,
        Result: resultChan,
        Error:  errorChan,
    }

    select {
    case wp.jobQueue <- job:
        // Trabajo encolado exitosamente.
        log.Printf("Job %d encolado", job.ID)
    default:
        // Cola llena. Rechazamos el trabajo inmediatamente.
        return "", fmt.Errorf("servidor sobrecargado, intenta más tarde")
    }

    // Esperamos el resultado del worker.
    select {
    case res := <-resultChan:
        return res, nil
    case err := <-errorChan:
        return "", err
    }
}

// Stop cierra el pool de manera ordenada.
func (wp *WorkerPool) Stop() {
    close(wp.jobQueue)
    wp.wg.Wait()
    log.Println("WorkerPool detenido")
}

// --- USO EN UN HANDLER HTTP ---
var pool = NewWorkerPool(5, 100) // 5 workers, cola de 100 jobs.

func intensiveTaskHandler(w http.ResponseWriter, r *http.Request) {
    // Simulamos una tarea que lleva tiempo (ej., procesar datos).
    task := func() (string, error) {
        time.Sleep(2 * time.Second) // Simula trabajo intensivo.
        return "Procesamiento completado para: " + r.URL.Query().Get("data"), nil
    }

    result, err := pool.Submit(task)
    if err != nil {
        http.Error(w, err.Error(), http.StatusServiceUnavailable)
        return
    }

    w.WriteHeader(http.StatusOK)
    w.Write([]byte(result))
}

func main() {
    r := mux.NewRouter()
    r.HandleFunc("/process", intensiveTaskHandler).Methods("GET")

    srv := &http.Server{
        Handler:      r,
        Addr:         ":8080",
        WriteTimeout: 15 * time.Second,
        ReadTimeout:  15 * time.Second,
    }

    log.Println("Servidor iniciado en :8080")
    defer pool.Stop() // Asegura la parada ordenada al final.
    log.Fatal(srv.ListenAndServe())
}

Código en acción: Implementación de Rate Limiting por IP con Middleware

Este ejemplo implementa un rate limiter simple en memoria, por dirección IP, usando el algoritmo de "token bucket". Es ideal para demostración y para límites por instancia de servicio.

// rate_limiter_middleware.go
package main

import (
    "net/http"
    "sync"
    "time"

    "github.com/gorilla/mux"
)

// RateLimiter almacena el estado para una IP.
type ipRateLimiter struct {
    tokens      int
    lastUpdated time.Time
    mu          sync.Mutex
}

// RateLimitMiddleware configura el límite.
func RateLimitMiddleware(requestsPerMinute int, burst int) mux.MiddlewareFunc {
    // Mapa en memoria. En producción, usa Redis para múltiples instancias.
    ipStore := make(map[string]*ipRateLimiter)
    var storeMu sync.Mutex

    // Función para obtener el limiter de una IP.
    getLimiter := func(ip string) *ipRateLimiter {
        storeMu.Lock()
        defer storeMu.Unlock()

        limiter, exists := ipStore[ip]
        if !exists {
            limiter = &ipRateLimiter{
                tokens:      burst, // Comienza con el "burst" permitido.
                lastUpdated: time.Now(),
            }
            ipStore[ip] = limiter
        }
        return limiter
    }

    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            ip := r.RemoteAddr // Simplificado. En producción, usa X-Forwarded-For.
            limiter := getLimiter(ip)

            limiter.mu.Lock()
            defer limiter.mu.Unlock()

            // Reabastecimiento de tokens: añade tokens según el tiempo pasado.
            now := time.Now()
            elapsed := now.Sub(limiter.lastUpdated)
            tokensToAdd := int(elapsed.Seconds()) * requestsPerMinute / 60

            if tokensToAdd > 0 {
                limiter.tokens += tokensToAdd
                if limiter.tokens > burst {
                    limiter.tokens = burst
                }
                limiter.lastUpdated = now
            }

            // Consume un token para esta petición.
            if limiter.tokens <= 0 {
                // Límite excedido.
                w.Header().Set("Retry-After", "60")
                http.Error(w, "Límite de tasa excedido", http.StatusTooManyRequests)
                return
            }

            limiter.tokens--
            // Pasa al siguiente handler (la ruta original).
            next.ServeHTTP(w, r)
        })
    }
}

// --- USO EN EL ROUTER ---
func main() {
    r := mux.NewRouter()

    // Aplica el middleware a todas las rutas.
    // Límite: 10 peticiones por minuto, con un burst de 5.
    r.Use(RateLimitMiddleware(10, 5))

    r.HandleFunc("/api/resource", func(w http.ResponseWriter, r *http.Request) {
        w.Write([]byte("¡Recurso accedido!"))
    }).Methods("GET")

    r.HandleFunc("/api/another", func(w http.ResponseWriter, r *http.Request) {
        w.Write([]byte("Otro endpoint"))
    }).Methods("GET")

    http.ListenAndServe(":8080", r)
}

Errores comunes y cómo evitarlos

1. Tamaño del pool o cola incorrecto: Un pool con demasiados workers puede saturar la base de datos o la CPU. Uno con muy pocos crea cuellos de botella. Una cola muy grande enmascara problemas de rendimiento y consume memoria; una muy pequeña rechaza trabajos prematuramente. Solución: Perfila tu aplicación bajo carga. Comienza con un número de workers igual al número de núcleos de CPU para tareas CPU-bound, o mayor para tareas I/O-bound. Monitoriza la longitud de la cola y los tiempos de espera.

2. Rate limiting en memoria para arquitecturas distribuidas: El ejemplo anterior usa un mapa en memoria. Si tienes múltiples instancias de tu servicio detrás de un balanceador de carga, cada instancia llevará su propio contador, permitiendo a un cliente superar el límite global multiplicado por el número de instancias. Solución: Para límites globales, usa un almacén compartido y de baja latencia como Redis con sus estructuras de datos atómicas (e.g., INCR con EXPIRE).

3. Bloquear en la goroutine principal al enviar trabajos: Enviar un trabajo a un canal del pool sin un select con timeout o sin manejar el caso de canal lleno puede hacer que un handler HTTP se bloquee indefinidamente si el pool está saturado. Solución: Como se muestra en el código, usa select con un caso default para rechazar trabajos inmediatamente cuando la cola esté llena, devolviendo un error 503.

4. No limpiar estado de rate limiting: El mapa en memoria de límites por IP crecerá indefinidamente, causando una fuga de memoria. Solución: Implementa un "barrido" periódico (un time.Ticker) que elimine las entradas de IPs que no han hecho peticiones en un largo período (e.g., 24 horas). En Redis, usa TTLs automáticos.

5. Aplicar rate limiting de forma demasiado agresiva o genérica: Limitar todos los endpoints por igual es injusto. Un endpoint de login debe tener un límite más estricto que uno de consulta de datos públicos. Solución: Implementa middlewares con diferentes configuraciones y aplícalos de forma granular a rutas o grupos de rutas en gorilla/mux usando router.PathPrefix().Subrouter().Use().

Checklist de dominio

Antes de considerar esta lección dominada, asegúrate de poder verificar los siguientes puntos:

• Puedo explicar la diferencia conceptual entre un Worker Pool (control de concurrencia) y Rate Limiting (control de rendimiento/throughput).
• He implementado un Worker Pool funcional que acepta tareas genéricas, las encola y procesa con un número fijo de goroutines.
• Sé integrar un Worker Pool en un handler HTTP de gorilla/mux para descargar tareas intensivas sin bloquear el hilo principal de la petición.
• Puedo escribir un middleware de rate limiting para gorilla/mux que limite peticiones por IP usando un algoritmo como token bucket.
• Comprendo la limitación del rate limiting en memoria y sé cuándo es necesario migrar a una solución distribuida (e.g., Redis).
• Sé manejar correctamente el caso en que un Worker Pool está saturado (cola llena), devolviendo un código HTTP 503 apropiado.
• Puedo decidir cuándo aplicar rate limiting global, por ruta o por cliente en función de los requisitos de negocio y seguridad.
• He probado ambos patrones bajo carga simulada (con herramientas como wrk o vegeta) y he interpretado los resultados.

Conclusión y siguiente paso

Los patrones de Worker Pool y Rate Limiting son herramientas esenciales en la caja de herramientas del desarrollador de APIs de alto rendimiento en Go. El primero te da el control sobre el consumo interno de recursos, mientras que el segundo te da control sobre la demanda externa. Juntos, te permiten construir microservicios que no solo son rápidos en condiciones ideales, sino también estables, predecibles y justos bajo una carga pesada o ataques de denegación de servicio.

El siguiente paso natural es llevar estos conceptos a un entorno de producción real. Esto implica integrar métricas y observabilidad: exponer métricas de la longitud de la cola del Worker Pool, de las tasas de rechazo y de los límites excedidos a sistemas como Prometheus. También implica considerar patrones más avanzados como circuit breakers y backpressure para sistemas altamente resilientes. Recuerda, la optimización para alto rendimiento es un viaje iterativo de medición, ajuste y aprendizaje continuo.