Colas de tareas en Go: creación, procesamiento y optimización

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

En el desarrollo moderno de backend, a menudo surgen tareas que no requieren una respuesta inmediata al usuario: enviar correos electrónicos, generar informes, procesar imágenes, sincronizar con APIs externas. Ejecutar estas operaciones de forma síncrona dentro de una solicitud HTTP es una mala práctica, ya que ralentiza la respuesta y aumenta la carga en el servidor.

Las colas de tareas (job queues) resuelven este problema. En Go (Golang), los mecanismos de concurrencia integrados — gorutinas y canales — permiten construir colas robustas y eficientes sin dependencias externas. En este artículo, analizaremos cómo crear una cola de tareas desde cero, optimizarla con un worker pool y evitar errores típicos.

Fundamentos: gorutinas y canales

Go proporciona dos primitivas clave para trabajar con colas:

• Gorutinas — hilos ligeros que se inician con la palabra clave go.
• Canales — «tuberías» tipadas para transferir datos entre gorutinas.

La cola más simple es una gorutina que lee de un canal y procesa tareas:

package main

import (    "fmt"    "time")

func worker(jobs <-chan int) {    for job := range jobs {        fmt.Printf("Procesando tarea %d", job)        time.Sleep(1 * time.Second) // simulación de trabajo    }}

func main() {    jobs := make(chan int, 10)

    // iniciamos el worker    go worker(jobs)

    // enviamos tareas    for i := 1; i <= 5; i++ {        jobs <- i    }

    close(jobs) // cerramos el canal después de enviar todas las tareas    time.Sleep(2 * time.Second)}

Este código crea un solo worker que procesa tareas secuencialmente. Pero para proyectos reales se necesita escalabilidad — varios workers.

Implementación de Worker Pool en Go

Worker pool (grupo de workers) es un patrón en el que iniciamos una cantidad fija de gorutinas workers, cada una tomando tareas de un canal común. Esto permite controlar el grado de paralelismo y la carga del sistema.

Implementación básica

package main

import (    "fmt"    "sync"    "time")

func worker(id int, jobs <-chan int, wg *sync.WaitGroup) {    defer wg.Done()    for job := range jobs {        fmt.Printf("Worker %d: tarea %d", id, job)        time.Sleep(500 * time.Millisecond)    }}

func main() {    const numWorkers = 3    const numJobs = 10

    jobs := make(chan int, numJobs)    var wg sync.WaitGroup

    // iniciamos los workers    for i := 1; i <= numWorkers; i++ {        wg.Add(1)        go worker(i, jobs, &wg)    }

    // enviamos tareas    for j := 1; j <= numJobs; j++ {        jobs <- j    }    close(jobs)

    wg.Wait() // esperamos a que todos los workers terminen    fmt.Println("Todas las tareas completadas")}

Aquí usamos sync.WaitGroup para esperar a que todos los workers terminen. El canal con búfer numJobs permite enviar todas las tareas sin bloqueo, mientras los workers no las hayan recogido.

Adición dinámica de tareas

En aplicaciones reales, las tareas llegan de forma asíncrona — por ejemplo, a través de HTTP. Aquí hay un ejemplo con un manejador:

package main

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

type Task struct {    ID   int    Data string}

var (    taskQueue = make(chan Task, 100)    wg        sync.WaitGroup)

func worker(id int) {    defer wg.Done()    for task := range taskQueue {        fmt.Printf("[Worker %d] Procesando: %s (ID=%d)", id, task.Data, task.ID)        time.Sleep(1 * time.Second)    }}

func handler(w http.ResponseWriter, r *http.Request) {    // obtenemos la tarea de la solicitud (simplificado)    task := Task{ID: time.Now().Nanosecond(), Data: "Nueva tarea"}    taskQueue <- task    w.W