PostgreSQL como job queue: ¿de verdad necesitas Redis o RabbitMQ?

Introducción

En este artículo veremos por qué tiene sentido usar Postgres como cola, analizaremos el diseño del esquema, implementaremos un polling de tareas seguro en concurrencia con SELECT … FOR UPDATE SKIP LOCKED y nos aseguraremos de que solo un worker procese una tarea concreta en cada momento. Al terminar tendrás el plano de un sistema fiable y transaccionalmente seguro para procesar tareas en segundo plano, todo dentro de Postgres.

Índice

1. ¿Por qué usar Postgres como job queue?
2. Conceptos clave y definiciones
3. Diseño del esquema de la base de datos
4. Bloqueos en Postgres: FOR UPDATE SKIP LOCKED
5. Insertar tareas nuevas
6. Polling de las tareas que hay que procesar
7. Actualizar el estado de las tareas
8. Workers en paralelo y concurrencia
9. Reintentos y gestión de errores
10. Ventajas e inconvenientes
11. Ejemplo de implementación paso a paso
12. Consejos prácticos de uso
13. Conclusión y una pregunta para reflexionar

1. ¿Por qué usar Postgres como job queue?

Cuando se habla de job queues, los desarrolladores suelen recurrir a herramientas como Redis, RabbitMQ o sistemas especializados como Sidekiq (en el ecosistema Ruby). Todas son opciones perfectamente válidas. Sin embargo, hay buenas razones para considerar Postgres para tus colas:

1. Simplicidad: si tu infraestructura ya se apoya en Postgres, no tienes que instalar ni configurar otro servicio. Así reduces el mantenimiento y el tiempo necesario para poner en marcha la cola.
2. Consistencia de los datos: Postgres es una base de datos robusta y conforme a ACID. Gestionar las tareas dentro de la misma base de datos garantiza una consistencia fuerte y la integridad transaccional.
3. Fiabilidad: si ya confías en Postgres para guardar los datos críticos de tu aplicación, también puedes confiarle la gestión de los trabajos en segundo plano. Evitas introducir varios puntos de fallo, porque todo está en un solo sitio.
4. Copias de seguridad sencillas: puedes hacer la copia de seguridad de la cola junto con los datos de la aplicación en un único proceso. No hace falta implementar ni mantener una estrategia de backup aparte para un sistema adicional.

Eso sí, conviene saber que Postgres puede no ser la solución de colas ideal en escenarios de escala enorme, con decenas de miles de trabajos publicados por segundo. Es muy capaz, pero siempre hay que medir el rendimiento y optimizar en consecuencia, o quizá valorar sistemas más especializados en los casos extremos. Para la mayoría de las cargas pequeñas y medianas, Postgres va de sobra.

🔔 Si necesitas una job queue con todavía más funcionalidades, pero más sencilla de usar y con una API REST limpia y simple, echa un vistazo a DMSContainer con su módulo Event Stream. Está listo para usar y con DMSContainer puedes empezar a usar colas, enviar correos, generar informes Excel y PDF (y mucho más) en minutos. ¡Pruébalo!

2. Conceptos clave y definiciones

Antes de entrar en detalles, fijemos algunos términos que vamos a usar:

• Job/Task: usaremos job y task (trabajo y tarea) indistintamente para referirnos a una unidad de trabajo que hay que procesar.
• Producer: una entidad (normalmente parte de tu aplicación) que crea tareas y las inserta en la cola.
• Consumer/Worker: un proceso en segundo plano que recoge de la cola las tareas pendientes, las procesa y actualiza su estado.
• FOR UPDATE SKIP LOCKED: una funcionalidad de Postgres que permite bloquear las filas a medida que se seleccionan, saltándose las que ya están bloqueadas por otra transacción. Es la clave para evitar conflictos cuando varios workers hacen polling de las tareas.

3. Diseño del esquema de la base de datos

Diseñar una tabla para guardar las tareas es bastante sencillo. Veamos un esquema mínimo que podrías usar:

CREATE TABLE tasks (
    id SERIAL PRIMARY KEY,
    status VARCHAR(50) NOT NULL DEFAULT 'pending',
    payload JSONB NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT now(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT now()
);

Esto es lo que hace cada columna:

• id: un identificador único para cada tarea, generado automáticamente.
• status: indica el estado de la tarea, por ejemplo ‘pending’, ‘in_progress’, ‘completed’ o ‘failed’.
• payload: contiene los datos necesarios para ejecutar la tarea. JSONB da flexibilidad para guardar estructuras de datos muy variadas.
• created_at: timestamp que marca cuándo se creó la tarea.
• updated_at: timestamp que se actualiza cada vez que cambia el estado de la tarea.

Mantendremos el esquema ligero. En la práctica podrías añadir índices sobre campos JSON concretos, u otras columnas como un nivel de prioridad o referencias a otras tablas.

4. Bloqueos en Postgres: FOR UPDATE SKIP LOCKED

La estrella del espectáculo es el bloqueo a nivel de fila de Postgres, en concreto SELECT … FOR UPDATE SKIP LOCKED. Introducido en PostgreSQL 9.5, SKIP LOCKED garantiza que, si varias transacciones intentan bloquear las mismas filas, una las bloquea y las demás se saltan las filas ya bloqueadas.

Esto significa que puedes ejecutar la misma sentencia SELECT en paralelo desde varios workers sin riesgo de duplicados. Cada worker toma una fila (es decir, una tarea) y la bloquea para que nadie más pueda tomarla. Es perfecto para repartir tareas entre varios workers, porque evita colisiones.

En resumen:

• FOR UPDATE: adquiere un bloqueo sobre la fila o las filas, de modo que ninguna otra transacción pueda modificarlas hasta que se libere el bloqueo.
• SKIP LOCKED: le dice a Postgres que no espere a las filas bloqueadas, sino que se las salte y bloquee solo las que no están bloqueadas en ese momento.

5. Insertar tareas nuevas

Cuando tu aplicación crea una tarea nueva, basta con insertar una fila en la tabla tasks. Un ejemplo sencillo:

INSERT INTO tasks (payload) 
VALUES ('{"task_type": "send_email", "recipient": "[email protected]"}');

El status vale ‘pending’ por defecto, created_at vale now() por defecto y updated_at también se fija en now(). Todo es DML estándar que ya conoces, sin sorpresas.

6. Polling de las tareas que hay que procesar

Ahora viene la parte importante: ¿cómo recogemos las tareas de forma que solo un worker procese una tarea dada en cada momento? Supongamos que tenemos varios procesos (o hilos) worker y que cada uno ejecuta una SELECT para encontrar la siguiente tarea pendiente.

Una versión simplificada:

BEGIN;

WITH cte_task AS (
    SELECT id
    FROM tasks
    WHERE status = 'pending'
    ORDER BY id
    FOR UPDATE SKIP LOCKED
    LIMIT 1
)
UPDATE tasks
SET status = 'in_progress',
    updated_at = now()
FROM cte_task
WHERE tasks.id = cte_task.id
RETURNING tasks.id, tasks.payload;

Hacemos todo lo anterior en una sola transacción:

1. WITH cte_task: esta common table expression bloquea exactamente una tarea en estado ‘pending’. Usamos ORDER BY id solo para tener un orden determinista, aunque podrías ordenar por fecha de creación o por prioridad si lo prefieres.
2. FOR UPDATE SKIP LOCKED: bloquea la fila y se la salta si otro worker ya la ha bloqueado.
3. LIMIT 1: garantiza que bloqueamos una sola tarea cada vez.
4. UPDATE: cuando la CTE encuentra la fila, cambiamos su estado a in_progress, con lo que asignamos la tarea a este worker.
5. RETURNING: nos devuelve el id y el payload de la tarea con la que vamos a trabajar.

Si la consulta anterior devuelve cero filas, significa que no hay tareas en estado ‘pending’. El worker puede hacer commit y dormir un rato antes de volver a intentarlo.

Si devuelve una fila, el worker puede procesar la tarea. Al terminar el trabajo, el worker puede lanzar otra UPDATE para marcar la tarea como ‘completed’, o como ‘failed’ si el procesamiento ha dado un error.

Este enfoque garantiza que cada tarea se bloquea y se actualiza en un paso atómico, e impide que otros workers tomen la misma tarea al mismo tiempo.

7. Actualizar el estado de las tareas

Cuando un worker termina la tarea, tiene que actualizar la tabla tasks en consecuencia:

UPDATE tasks
SET status = 'completed',
    updated_at = now()
WHERE id = <task_id>;

O, si el procesamiento de la tarea falla, podrías hacer:

UPDATE tasks
SET status = 'failed',
    updated_at = now()
WHERE id = <task_id>;

Quizá también quieras guardar mensajes de error, el número de reintentos u otros metadatos relevantes. Plantéate añadir estos campos al esquema si prevés que tus tareas pueden fallar y necesitar un nuevo procesamiento.

8. Workers en paralelo y concurrencia

Usar FOR UPDATE SKIP LOCKED significa que varios workers pueden ejecutar la misma consulta de polling al mismo tiempo sin conflictos. Cada worker acaba bloqueando filas distintas. Si un worker bloquea una fila, los demás se la saltan. Este enfoque de concurrencia es ideal para repartir tareas entre un pool de workers.

Sin embargo, tienes que controlar con qué frecuencia hace polling cada worker. Un polling demasiado frecuente puede cargar mucho la base de datos, sobre todo si tienes muchos procesos worker. Hay que encontrar el equilibrio entre recoger las tareas rápido y no machacar la base de datos con demasiadas SELECT.

9. Reintentos y gestión de errores

En el mundo real, el procesamiento de trabajos nunca está libre de errores al 100%. Necesitas una estrategia para las tareas que fallan. Supongamos que un worker toma una tarea, pero la ejecución falla por un error de red o por cualquier otro problema. Puedes:

1. Marcar la tarea como ‘failed’ y registrar los detalles del error. Más tarde, otro subsistema o un script específico puede buscar las tareas ‘failed’ y decidir si reintentarlas o no.
2. Reintentar automáticamente un número limitado de veces. Puedes llevar la cuenta en una columna retry_count de la tabla tasks. Si está por debajo de un umbral, devuelves la tarea al estado ‘pending’ e incrementas retry_count.

Un ejemplo que marca una tarea como fallida e incrementa retry_count:

ALTER TABLE tasks ADD COLUMN retry_count INT NOT NULL DEFAULT 0;

BEGIN;

UPDATE tasks
SET status = 'failed',
    retry_count = retry_count + 1,
    updated_at = now()
WHERE id = <task_id>;

COMMIT;

A partir de ahí puedes tener un proceso aparte o un cron job que busque las tareas failed con retry_count < 5 y las devuelva a ‘pending’, es decir, que las vuelva a encolar para otro intento. En escenarios más sencillos, el propio worker puede poner retry_count a retry_count + 1 y dejar de procesar la tarea después de un número de intentos determinado.

10. Ventajas e inconvenientes

Aunque usar Postgres como job queue puede ser muy cómodo, no es una solución universal. Esta lista de ventajas e inconvenientes refleja la opinión general entre los desarrolladores:

Ventajas

• Sin infraestructura adicional: ya tienes Postgres, así que no hay nada más que instalar ni mantener.
• Seguridad transaccional: si tu código está muy ligado a los datos guardados en Postgres, tener las tareas en el mismo sitio simplifica la consistencia.
• Bloqueo a nivel de fila: Postgres tiene mecanismos de concurrencia robustos. FOR UPDATE SKIP LOCKED está muy probado y es fiable.
• Backup y restauración: todos los datos, tareas incluidas, se pueden respaldar juntos.

Inconvenientes

• Escalabilidad: si tienes un throughput altísimo o necesitas funcionalidades de cola avanzadas (enrutamiento avanzado, colas con prioridad, etc.), puede que te convengan sistemas especializados.
• Carga de la base de datos: el polling de tareas puede añadir carga a la base de datos principal. Puedes mitigarlo con una arquitectura bien pensada o con una réplica, pero es algo más que tener en cuenta.
• Funcionalidades limitadas: sistemas como RabbitMQ o Kafka ofrecen enrutamiento de mensajes sofisticado, fan-out y más. Postgres es más sencillo en ese aspecto, pero también menos flexible para ciertos patrones.

11. Ejemplo de implementación paso a paso

Veamos un escenario hipotético con una pequeña startup tecnológica: Acme Email Services. Se ocupan de correos transaccionales y necesitan encolar las tareas de envío para no sobrecargar su servidor SMTP.

1. Crear la tabla tasks:

   CREATE TABLE tasks (
       id SERIAL PRIMARY KEY,
       status VARCHAR(50) NOT NULL DEFAULT 'pending',
       payload JSONB NOT NULL,
       created_at TIMESTAMP WITH TIME ZONE DEFAULT now(),
       updated_at TIMESTAMP WITH TIME ZONE DEFAULT now()
   );
   

2. Insertar tareas (lado del producer, por ejemplo desde un endpoint de una API):

   INSERT INTO tasks (payload)
   VALUES ('{"task_type": "send_email", "subject": "Welcome!", "recipient": "[email protected]", "body": "Hello and welcome!"}');
   

3. Proceso worker (pseudocódigo en Python, por ejemplo):

   import psycopg2
   import time
   
   def poll_and_process_tasks():
       while True:
           conn = psycopg2.connect("dbname=acme user=postgres password=postgres")
           conn.autocommit = False
           try:
               with conn.cursor() as cur:
                   cur.execute("""
                       WITH cte_task AS (
                           SELECT id, payload
                           FROM tasks
                           WHERE status = 'pending'
                           ORDER BY id
                           FOR UPDATE SKIP LOCKED
                           LIMIT 1
                       )
                       UPDATE tasks
                       SET status = 'in_progress',
                           updated_at = now()
                       FROM cte_task
                       WHERE tasks.id = cte_task.id
                       RETURNING tasks.id, tasks.payload;
                   """)
                   row = cur.fetchone()
                   if row:
                       task_id, task_payload = row
                       # Procesa la tarea
                       process_email(task_payload)  # tu función personalizada
   
                       # Marca como completada
                       cur.execute("""
                           UPDATE tasks
                           SET status = 'completed', updated_at = now()
                           WHERE id = %s
                       """, (task_id,))
                       conn.commit()
                   else:
                       conn.rollback()
                       # No hay tareas, duerme un poco antes de volver a mirar
                       time.sleep(5)
           finally:
               conn.close()
   
   def process_email(payload):
       # Pseudocódigo para enviar el correo
       # payload['task_type'] == 'send_email'
       # Usa payload['recipient'], payload['subject'], etc.
       print(f"Sending email to {payload['recipient']}...")
       # Aquí se envía el correo de verdad
   

   En el script anterior:

   • Nos conectamos a la base de datos y empezamos una transacción.
   • Intentamos obtener una sola tarea pending con la técnica FOR UPDATE SKIP LOCKED.
   • Si la conseguimos, la marcamos inmediatamente como in_progress.
   • Después procesamos el correo (omitimos el código real de envío por brevedad).
   • Si todo va bien, cambiamos su estado a completed.
   • Si no hay tareas pendientes, hacemos rollback (para liberar cualquier bloqueo o actualización parcial) y dormimos un poco.

⭐ La versión Delphi completa del sistema de job queue se publicará para los suscriptores de PATREON en los próximos días.

4. Gestión de errores: si se produce un error al enviar el correo, puedes capturar la excepción, marcar la tarea como failed y, si quieres, guardar el error. Después puedes aplicar una estrategia de reencolado si hace falta.

12. Consejos prácticos de uso

• Índices: si tienes un volumen de tareas muy alto, plantéate crear un índice sobre (status) o (status, id) para acelerar la búsqueda de las tareas pendientes.
• Limita la frecuencia de polling: implementa una pequeña espera en los workers o una estrategia de backoff para reducir la carga de la base de datos.
• Usa una tabla aparte: si tu aplicación tiene varios tipos de tareas o un número enorme de ellas, puedes separarlas en tablas distintas o incluso en esquemas distintos. Ayuda a organizar y a afinar el rendimiento.
• Monitorización: vigila el número de tareas en cada estado. Herramientas como Grafana o métricas propias pueden avisarte si empieza a formarse un atasco (por ejemplo, si tus tareas se quedan en ‘pending’ demasiado tiempo).
• Tamaño de las transacciones: presta atención a los límites de las transacciones. Si intentas bloquear y procesar cientos de tareas en una sola transacción, podrías mantener los bloqueos demasiado tiempo. Normalmente conviene procesar las tareas de una en una o en lotes pequeños.

13. Conclusión y una pregunta para reflexionar

A estas alturas has visto que usar Postgres como job queue puede ser práctico y elegante, sobre todo si tu entorno ya usa Postgres para otras cosas. Evitas la complejidad de sistemas adicionales, obtienes garantías transaccionales sólidas y aprovechas las funcionalidades de concurrencia de Postgres para repartir tareas entre varios workers sin duplicados.

Con PostgreSQL puedes montar un clúster de procesos worker que gestione de forma fiable miles de tareas al día. Guarda todos los logs y los metadatos de los trabajos en Postgres, y consultar datos históricos, analizar el rendimiento y gestionar la recuperación de errores será sencillo. Aunque soluciones como DMSContainer/EventStream!, Redis, RabbitMQ o Kafka pueden encajar mejor en ciertos escenarios de gran escala o especializados, el enfoque de job queue con Postgres es un candidato sólido para muchas necesidades de escala pequeña y mediana.

Pregunta para reflexionar: con la carga de trabajo y la infraestructura de tu organización, ¿necesitas las funcionalidades especializadas de un sistema de colas dedicado, o puede Postgres procesar tus trabajos lo bastante bien como para simplificar tu arquitectura? Escríbenos un correo si necesitas consultoría y desarrollo especializados.

Artículos relacionados sobre PostgreSQL:

• Tipos compuestos en PostgreSQL: una guía completa - Modelado de datos avanzado para colas complejas
• IDENTITY vs SERIAL en PostgreSQL - Buenas prácticas para las primary key
• Cómo acelerar las consultas con LIKE usando pg_trgm - Técnicas de optimización de consultas

Referencias

• Documentación oficial de PostgreSQL:
  https://www.postgresql.org/docs/current/sql-select.html#SQL-FOR-UPDATE-SHARE
  Detalles sobre el uso y la sintaxis de FOR UPDATE SKIP LOCKED.

• Wiki de PostgreSQL sobre colas:
  https://wiki.postgresql.org/wiki/Category:Queueing
  Discusiones de la comunidad y extensiones para implementar colas en Postgres.

• Propiedades ACID:
  https://en.wikipedia.org/wiki/ACID
  Explicación de las garantías transaccionales que Postgres ofrece de serie.

Y esta es la historia. Tanto si estás construyendo un proyecto personal, mejorando una herramienta interna o necesitas una solución rápida pero fiable para el procesamiento en segundo plano, Postgres puede ser el héroe de tu job queue. Con un esquema bien diseñado, la magia de FOR UPDATE SKIP LOCKED y un poco de lógica en los workers, tendrás un sistema que mantiene tus tareas ordenadas, tus workers ocupados y tu infraestructura simplificada. Disfruta de la sencillez y la fiabilidad de tu nuevo sistema de colas, movido por el bueno de Postgres.

¿Necesitas más?

Únete a para apoyar el proyecto y acceder a contenido premium como artículos, vídeos y otros análisis. Recuerda unirte a la comunidad de PATREON para recibir información útil, tutoriales y análisis, tener soporte prioritario y mucho más. También puedes apoyar el proyecto con Buy Me a Coffe y obtener las mismas ventajas.

¡Que lo disfrutes!

– Daniele Teti