Colas de tareas en Go: guía práctica

Introducción: ¿por qué se necesitan colas de tareas en Go?

En el desarrollo moderno, a menudo surge la necesidad de procesar tareas de forma asíncrona, sin bloquear el hilo principal de la aplicación. Las colas de tareas (task queues) son un patrón clásico que permite distribuir el trabajo entre varios procesadores (workers), gestionar la carga y garantizar la tolerancia a fallos.

El lenguaje Go (Golang) proporciona mecanismos integrados para implementar colas de tareas: gorutinas (goroutines) y canales (channels). En este artículo, analizaremos en la práctica cómo construir una cola de tareas confiable utilizando la biblioteca estándar de Go, sin recurrir a intermediarios de mensajes externos (aunque en producción RabbitMQ o Redis pueden ser útiles).

Parte principal

1. Patrón básico de Worker Pool

Worker Pool es un conjunto de un número fijo de gorutinas (workers) que escuchan un solo canal y procesan las tareas entrantes. Esto permite controlar el grado de paralelismo y evitar la sobrecarga del sistema.

Consideremos un ejemplo simple: creamos un pool de 3 workers que procesan números enteros (simulación de trabajo).

package main\n\nimport (\n    "fmt"\n    "sync"\n    "time"\n)\n\nfunc worker(id int, jobs <-chan int, wg *sync.WaitGroup) {\n    defer wg.Done()\n    for job := range jobs {\n        fmt.Printf("Worker %d inició la tarea %d\\n", id, job)\n        time.Sleep(time.Second) // simulación de trabajo\n        fmt.Printf("Worker %d completó la tarea %d\\n", id, job)\n    }\n}\n\nfunc main() {\n    const numJobs = 5\n    const numWorkers = 3\n\n    jobs := make(chan int, numJobs)\n    var wg sync.WaitGroup\n\n    // Iniciamos los workers\n    for w := 1; w <= numWorkers; w++ {\n        wg.Add(1)\n        go worker(w, jobs, &wg)\n    }\n\n    // Enviamos las tareas\n    for j := 1; j <= numJobs; j++ {\n        jobs <- j\n    }\n    close(jobs) // cerramos el canal para que los workers finalicen\n\n    wg.Wait()\n    fmt.Println("Todas las tareas completadas")\n}

Análisis del código:

\n
• jobs — canal con búfer al que enviamos las tareas.
\n
• worker — función que lee del canal y procesa cada tarea.
\n
• sync.WaitGroup — espera a que todos los workers finalicen.
\n
• Después de enviar todas las tareas, el canal se cierra (close(jobs)), lo que provoca la finalización del bucle for range en los workers.
\n

Este patrón es la base para soluciones más complejas.

2. Cola con prioridades y tiempos de espera

En aplicaciones reales, las tareas pueden tener diferente prioridad o requerir ejecución dentro de un tiempo determinado. Implementemos una cola donde las tareas con prioridad se procesen primero, y agreguemos un contexto con tiempo de espera para cancelar tareas largas.

package main\n\nimport (\n    "context"\n    "fmt"\n    "time"\n)\n\ntype Task struct {\n    ID       int\n    Priority int // cuanto menor es el número, mayor es la prioridad\n    Data     string\n}\n\nfunc processTask(ctx context.Context, task Task) error {\n    select {\n    case <-ctx.Done():\n        return ctx.Err()\n    case <-time.After(time.Duration(task.Priority) * 500 * time.Millisecond):\n        fmt.Printf("Procesada tarea %d: %s (prioridad %d)\\n", task.ID, task.Data, task.Priority)\n        return nil\n    }\n}\n\nfunc main() {\n    tasks := []Task{\n        {ID: 1, Priority: 3, Data: "Prioridad baja"},\n        {ID: 2, Priority: 1, Data: "Prioridad alta"},\n        {ID: 3, Priority: 2, Data: "Prioridad media"},\n    }\n\n    // Ordenamiento por prioridad (las más altas primero)\n    // En un proyecto real, use ordenamiento o un heap (container/heap)\n    for i := 0; i < len(tasks); i++ {\n        for j := i + 1; j < len(tasks); j++ {\n            if tasks[j].Priority < tasks[i].Priority {\n                tasks[i], tasks[j] = tasks[j], tasks[i]\n            }\n