PostgreSQL Job Queue: SKIP LOCKED, particionamiento y monitoreo sin Redis

Introducción: cuándo PostgreSQL supera a Redis o a un broker de mensajes

La mayoría de los equipos recurren por defecto a Redis o RabbitMQ cuando se trata de colas de tareas. Sin embargo, en 2026, PostgreSQL es una plataforma madura y battle-tested con garantías transaccionales que Redis no ofrece de fábrica. Si ya tienes PostgreSQL en tu stack, añadir un broker independiente implica: un nuevo punto de fallo, costes operativos adicionales, mayor complejidad en la infraestructura y la necesidad de mantener la coherencia entre dos almacenes de datos.

Una cola en PostgreSQL está justificada en los siguientes escenarios:

Las tareas están estrechamente ligadas a los datos de negocio y requieren operaciones atómicas con la base de datos principal.
La carga es moderada: hasta unos pocos miles de tareas por segundo.
Es importante garantizar semántica exactly-once o at-least-once con confirmación transaccional.
El equipo quiere simplificar el stack y evitar la complejidad operativa de Redis Cluster.

En este artículo construiremos una cola de tareas lista para producción sin Redis: con SKIP LOCKED, particionamiento, workers en Go y PHP/Laravel, monitoreo mediante Grafana y despliegue en Kubernetes.

Patrón Job Queue en PostgreSQL: esquema, estados e índices

La base del patrón es una tabla de trabajos con estados explícitos y bloqueo pesimista. Aquí está el esquema base:

CREATE TABLE jobs (
    id              BIGSERIAL PRIMARY KEY,
    queue           TEXT NOT NULL DEFAULT 'default',
    payload         JSONB NOT NULL,
    status          TEXT NOT NULL DEFAULT 'pending'
                        CHECK (status IN ('pending','processing','done','failed')),
    attempts        INT NOT NULL DEFAULT 0,
    max_attempts    INT NOT NULL DEFAULT 3,
    scheduled_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    locked_until    TIMESTAMPTZ,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
) PARTITION BY LIST (queue);

CREATE TABLE jobs_default PARTITION OF jobs FOR VALUES IN ('default');
CREATE TABLE jobs_email   PARTITION OF jobs FOR VALUES IN ('email');
CREATE TABLE jobs_reports PARTITION OF jobs FOR VALUES IN ('reports');

-- Índices para un polling eficiente
CREATE INDEX idx_jobs_default_pending
    ON jobs_default (scheduled_at)
    WHERE status = 'pending';

CREATE INDEX idx_jobs_default_processing
    ON jobs_default (locked_until)
    WHERE status = 'processing';

El campo scheduled_at permite implementar tareas diferidas. locked_until se utiliza para detectar workers bloqueados: si un worker cae, la tarea vuelve automáticamente a la cola al expirar el TTL del bloqueo. El campo payload de tipo JSONB aporta flexibilidad sin necesidad de modificar el esquema para cada tipo de tarea.

SELECT ... FOR UPDATE SKIP LOCKED: el mecanismo clave

El principal problema de las colas sobre SQL es el acceso concurrente de varios workers a la misma tarea. El enfoque clásico con SELECT + UPDATE en dos consultas genera una race condition. SELECT ... FOR UPDATE SKIP LOCKED, introducido en PostgreSQL 9.5, resuelve esto de forma elegante y atómica.

Principio de funcionamiento: cuando un worker ejecuta SELECT ... FOR UPDATE, PostgreSQL bloquea la fila. Otros workers que ejecuten la misma consulta con SKIP LOCKED simplemente omiten las filas bloqueadas en lugar de esperar a que se libere el bloqueo. Esto convierte a PostgreSQL en un eficiente gestor de bloqueos distribuidos para la cola.

-- Captura atómica de una tarea por el worker
WITH next_job AS (
    SELECT id
    FROM jobs
    WHERE queue = 'default'
      AND status = 'pending'
      AND scheduled_at <= NOW()
    ORDER BY scheduled_at
    LIMIT 1
    FOR UPDATE SKIP LOCKED
)
UPDATE jobs
SET
    status       = 'processing',
    attempts     = attempts + 1,
    locked_until = NOW() + INTERVAL '5 minutes',
    updated_at   = NOW()
FROM next_job
WHERE jobs.id = next_job.id
RETURNING jobs.*;

Toda la consulta se ejecuta en una sola transacción. Si el worker no confirma la ejecución de la tarea antes de locked_until, un proceso reaper independiente devuelve la tarea al estado pending:

-- Recuperación de tareas bloqueadas (reaper)
UPDATE jobs
SET status = 'pending', updated_at = NOW()
WHERE status = 'processing'
  AND locked_until < NOW()
  AND attempts < max_attempts;

Implementación del worker en Go: polling, procesamiento y confirmación

Go es ideal para escribir workers: bajo consumo de memoria, goroutines para procesamiento paralelo y context integrado para el graceful shutdown. Utilizamos pgx como driver de PostgreSQL.

package main

import (
    "context"
    "encoding/json"
    "log"
    "time"

    "github.com/jackc/pgx/v5/pgxpool"
)

type Job struct {
    ID      int64
    Queue   string
    Payload json.RawMessage
}

func acquireJob(ctx context.Context, pool *pgxpool.Pool, queue string) (*Job, error) {
    row := pool.QueryRow(ctx, `
        WITH next_job AS (
            SELECT id FROM jobs
            WHERE queue = $1
              AND status = 'pending'
              AND scheduled_at <= NOW()
            ORDER BY scheduled_at
            LIMIT 1
            FOR UPDATE SKIP LOCKED
        )
        UPDATE jobs
        SET status = 'processing',
            attempts = attempts + 1,
            locked_until = NOW() + INTERVAL '5 minutes',
            updated_at = NOW()
        FROM next_job
        WHERE jobs.id = next_job.id
        RETURNING jobs.id, jobs.queue, jobs.payload
    `, queue)

    var job Job
    err := row.Scan(&job.ID, &job.Queue, &job.Payload)
    if err != nil {
        return nil, err
    }
    return &job, nil
}

func completeJob(ctx context.Context, pool *pgxpool.Pool, id int64) error {
    _, err := pool.Exec(ctx,
        `UPDATE jobs SET status = 'done', updated_at = NOW() WHERE id = $1`, id)
    return err
}

func failJob(ctx context.Context, pool *pgxpool.Pool, id int64) error {
    _, err := pool.Exec(ctx, `
        UPDATE jobs
        SET status = CASE WHEN attempts >= max_attempts THEN 'failed' ELSE 'pending' END,
            updated_at = NOW()
        WHERE id = $1
    `, id)
    return err
}

func runWorker(ctx context.Context, pool *pgxpool.Pool, queue string) {
    for {
        select {
        case <-ctx.Done():
            return
        default:
        }

        job, err := acquireJob(ctx, pool, queue)
        if err != nil {
            // Sin tareas — esperamos antes del siguiente poll
            time.Sleep(500 * time.Millisecond)
            continue
        }

        log.Printf("Processing job %d", job.ID)
        if err := processPayload(job.Payload); err != nil {
            log.Printf("Job %d failed: %v", job.ID, err)
            _ = failJob(ctx, pool, job.ID)
            continue
        }
        _ = completeJob(ctx, pool, job.ID)
        log.Printf("Job %d done", job.ID)
    }
}

func processPayload(payload json.RawMessage) error {
    // Lógica de negocio para el procesamiento de la tarea
    return nil
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    pool, _ := pgxpool.New(ctx, "postgres://user:pass@localhost/db")
    defer pool.Close()

    // Lanzamos varios workers en paralelo
    for i := 0; i < 5; i++ {
        go runWorker(ctx, pool, "default")
    }

    // Graceful shutdown por señal...
    select {}
}

Nótese que, cuando acquireJob falla por ausencia de tareas, el worker realiza un breve sleep en lugar de un busy-wait. En producción, este valor se puede externalizar a la configuración y aplicar exponential backoff.

Implementación del worker en PHP/Laravel: driver personalizado

Laravel tiene soporte integrado para PostgreSQL a través del driver database, pero utiliza advisory locks en lugar de SKIP LOCKED. Escribiremos un driver personalizado que use el mecanismo de bloqueo correcto.

<?php

namespace App\Queue;

use Illuminate\Queue\DatabaseQueue;
use Illuminate\Queue\Jobs\DatabaseJob;
use Illuminate\Support\Facades\DB;

class SkipLockedQueue extends DatabaseQueue
{
    public function pop($queue = null)
    {
        $queue = $this->getQueue($queue);

        return $this->getConnection()->transaction(function () use ($queue) {
            $job = $this->getNextAvailableJobWithSkipLocked($queue);

            if ($job !== null) {
                return new DatabaseJob(
                    $this->container,
                    $this,
                    $job,
                    $this->connectionName,
                    $queue
                );
            }
        });
    }

    protected function getNextAvailableJobWithSkipLocked($queue)
    {
        $job = DB::selectOne("
            WITH next_job AS (
                SELECT id FROM jobs
                WHERE queue = ?
                  AND status = 'pending'
                  AND scheduled_at <= NOW()
                ORDER BY scheduled_at
                LIMIT 1
                FOR UPDATE SKIP LOCKED
            )
            UPDATE jobs
            SET status = 'processing',
                attempts = attempts + 1,
                locked_until = NOW() + INTERVAL '5 minutes',
                updated_at = NOW()
            FROM next_job
            WHERE jobs.id = next_job.id
            RETURNING jobs.*
        ", [$queue]);

        return $job ? (object) $job : null;
    }
}

Registramos el driver en AppServiceProvider:

<?php

// En AppServiceProvider::boot()
Queue::extend('pgsql_skip_locked', function () {
    return new SkipLockedQueueConnector();
});

En config/queue.php añadimos la conexión:

'pgsql_skip' => [
    'driver'    => 'pgsql_skip_locked',
    'table'     => 'jobs',
    'queue'     => 'default',
    'retry_after' => 300,
],

Ahora php artisan queue:work --queue=default --connection=pgsql_skip utiliza el SKIP LOCKED nativo de PostgreSQL en lugar de advisory locks.

Particionamiento de la tabla de cola: escalado y limpieza

Una tabla de cola sin limpieza crece indefinidamente. El particionamiento por queue (LIST partitioning, como en el esquema anterior) resuelve dos problemas: el aislamiento de carga entre colas y la limpieza eficiente mediante DROP PARTITION en lugar de DELETE.

Para limpiar las tareas completadas, añadimos particionamiento por tiempo con un esquema de archivo:

CREATE TABLE jobs_archive (
    LIKE jobs INCLUDING ALL
) PARTITION BY RANGE (created_at);

CREATE TABLE jobs_archive_2025_q1
    PARTITION OF jobs_archive
    FOR VALUES FROM ('2025-01-01') TO ('2025-04-01');

CREATE TABLE jobs_archive_2025_q2
    PARTITION OF jobs_archive
    FOR VALUES FROM ('2025-04-01') TO ('2025-07-01');

-- Movemos las tareas completadas al archivo (cron-job)
INSERT INTO jobs_archive
SELECT * FROM jobs
WHERE status IN ('done', 'failed')
  AND updated_at < NOW() - INTERVAL '7 days';

DELETE FROM jobs
WHERE status IN ('done', 'failed')
  AND updated_at < NOW() - INTERVAL '7 days';

La eliminación de una partición de archivo antigua es instantánea y no genera carga:

-- Eliminación del trimestre sin table lock
DROP TABLE jobs_archive_2025_q1;

Para escenarios de alta carga, se puede dividir la tabla principal por queue + rango de id, o utilizar pg_partman para la creación automática de particiones.

Monitoreo: métricas, Prometheus y Grafana

Sin monitoreo, la cola es una caja negra. Las métricas clave para una cola en PostgreSQL son:

Queue depth — número de tareas en estado pending por cada cola.
Processing time — tiempo medio y p95 de procesamiento de una tarea.
Failed rate — porcentaje de tareas en estado failed en los últimos N minutos.
Stuck jobs — tareas en processing con locked_until expirado.

Consultas SQL para recopilar métricas (exporter en Go o pg_stat_statements):

-- Profundidad de cola por tipo
SELECT queue, status, COUNT(*) as count
FROM jobs
GROUP BY queue, status;

-- Tiempo medio de procesamiento (últimos 10 minutos)
SELECT queue,
       AVG(EXTRACT(EPOCH FROM (updated_at - created_at))) AS avg_processing_sec,
       PERCENTILE_CONT(0.95) WITHIN GROUP (
           ORDER BY EXTRACT(EPOCH FROM (updated_at - created_at))
       ) AS p95_processing_sec
FROM jobs
WHERE status = 'done'
  AND updated_at > NOW() - INTERVAL '10 minutes'
GROUP BY queue;

-- Tareas bloqueadas
SELECT COUNT(*) AS stuck_count
FROM jobs
WHERE status = 'processing'
  AND locked_until < NOW();

Recopilamos las métricas mediante un exporter personalizado de Prometheus y las visualizamos en Grafana. Ejemplo de configuración del dashboard: panel con gráfico de Queue Depth over Time, alerta cuando pending > 1000 durante más de 5 minutos, y heatmap de distribución del tiempo de procesamiento por cola.

Para la integración con Prometheus utilizamos postgres_exporter con consultas personalizadas a través de queries.yaml:

pg_job_queue_depth:
  query: |
    SELECT queue, status, COUNT(*) as count
    FROM jobs GROUP BY queue, status
  metrics:
    - queue:
        usage: LABEL
    - status:
        usage: LABEL
    - count:
        usage: GAUGE
        description: Number of jobs by queue and status

Comparación con Redis Queues: ventajas y desventajas

La cola en PostgreSQL no es una solución universal. Aquí hay una comparación honesta:

Ventajas de PostgreSQL:

Transaccionalidad: la tarea y los datos de negocio se modifican atómicamente en una sola transacción.
Sin componente adicional en la infraestructura — menos puntos de fallo.
SQL completo para analítica, monitoreo y depuración.
Garantías de durabilidad (WAL) de fábrica, sin configuración adicional de Redis AOF/RDB.
Particionamiento e índices para gestionar eficientemente colas de gran tamaño.

Limitaciones de PostgreSQL:

Menor rendimiento que Redis: con cargas superiores a 10 000 tareas/seg, PostgreSQL empieza a quedarse atrás.
El polling genera carga en la base de datos; con polling agresivo, esto se nota en la CPU.
No hay pub/sub ni fan-out de fábrica — para eso se necesita un mecanismo separado (LISTEN/NOTIFY).
Carga de VACUUM: los UPDATE intensivos de estados generan dead tuples.

Conclusión: si tienes alta carga con miles de tareas por segundo, Redis/Kafka están justificados. Para la mayoría de las aplicaciones de negocio, la cola en PostgreSQL es una solución más sencilla, fiable y económica.

Despliegue de workers en Kubernetes: Deployment vs Job

En Kubernetes, los workers de cola se despliegan como Deployment (no como Job), ya que deben ejecutarse de forma continua y no realizar una tarea puntual. Ejemplo de manifiesto para un worker en Go:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: queue-worker
  namespace: production
spec:
  replicas: 3
  selector:
    matchLabels:
      app: queue-worker
  template:
    metadata:
      labels:
        app: queue-worker
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "9090"
    spec:
      containers:
      - name: worker
        image: myapp/queue-worker:1.2.0
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-secret
              key: url
        - name: QUEUE_NAME
          value: default
        - name: WORKER_CONCURRENCY
          value: "5"
        resources:
          requests:
            cpu: 100m
            memory: 64Mi
          limits:
            cpu: 500m
            memory: 256Mi
        livenessProbe:
          httpGet:
            path: /healthz
            port: 9090
          initialDelaySeconds: 5
          periodSeconds: 10
      terminationGracePeriodSeconds: 60

Aspectos clave del despliegue en Kubernetes:

terminationGracePeriodSeconds: damos al worker tiempo para completar la tarea en curso antes de detener el pod. El worker debe capturar SIGTERM y detener el polling tras finalizar la iteración actual.
HPA por métrica personalizada: configuramos el escalado horizontal basado en la profundidad de la cola a través del Prometheus Adapter — cuando las tareas pending aumentan, se añaden réplicas automáticamente.
PodDisruptionBudget: garantizamos un mínimo de 2 réplicas durante un rolling update.
Kubernetes Job se usa para el proceso reaper: lo ejecutamos según un calendario mediante CronJob cada 5 minutos para recuperar las tareas bloqueadas.

apiVersion: batch/v1
kind: CronJob
metadata:
  name: queue-reaper
spec:
  schedule: "*/5 * * * *"
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: reaper
            image: myapp/queue-reaper:1.0.0
            env:
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: db-secret
                  key: url
          restartPolicy: OnFailure

Conclusión

Construir una cola de tareas robusta en PostgreSQL es un objetivo perfectamente alcanzable sin recurrir a tecnologías adicionales. La combinación de SELECT ... FOR UPDATE SKIP LOCKED, un esquema bien diseñado con particionamiento, workers en Go o Laravel y monitoreo mediante Prometheus/Grafana proporciona una solución lista para producción para la mayoría de los casos de uso empresariales.

Una cola en PostgreSQL en 2026 es una elección arquitectónica consciente en favor de la simplicidad, la fiabilidad y la transaccionalidad. Empieza con una sola cola, mide la carga y solo si llegas al límite de rendimiento considera Redis o Kafka. Para el 80% de los proyectos, ese límite nunca llegará.

Construcción de una cola de tareas robusta en PostgreSQL: SKIP LOCKED, particionamiento y monitoreo sin Redis