Creando un Sistema de Gestión de Colas y Trabajos en Segundo Plano en PHP Puro

Introducción al Procesamiento en Segundo Plano con PHP 🚀

En el desarrollo web moderno, nos enfrentamos constantemente al desafío de ejecutar tareas pesadas que pueden ralentizar la experiencia del usuario. Acciones como enviar correos electrónicos masivos, procesar imágenes de alta resolución, generar reportes PDF complejos o interactuar con APIs externas lentas no deben bloquear el ciclo de vida HTTP de una solicitud. La clave para una aplicación rápida y escalable es delegar estas tareas a un sistema de colas (Queue Worker).

Aunque existen potentes herramientas en el ecosistema como Laravel Queue o RabbitMQ, entender cómo funciona este mecanismo desde los cimientos utilizando PHP puro te otorgará un control absoluto, optimizará tus recursos y te convertirá en un desarrollador más completo.

📋 ¿Qué es una Cola de Trabajos y por qué la necesitas?

Una cola de trabajos (Job Queue) sigue el principio FIFO (First In, First Out - Primero en entrar, primero en salir). Cuando ocurre un evento que requiere un procesamiento pesado, en lugar de ejecutarlo inmediatamente, creamos un objeto o estructura de datos llamada Job (trabajo) y lo depositamos en un almacenamiento temporal (Broker). Un proceso independiente llamado Worker se encarga de extraer estos trabajos de la cola y ejecutarlos en segundo plano.

🛠️ Arquitectura del Sistema de Colas

Para construir nuestro sistema en PHP puro, necesitaremos tres componentes fundamentales:

1. El Productor (Producer): Es tu aplicación web tradicional que intercepta la petición del usuario y empuja nuevos trabajos a la cola.
2. El Almacén (Broker): El medio físico donde se almacenan los trabajos pendientes. Usaremos una tabla en MySQL/MariaDB para este tutorial por su accesibilidad.
3. El Consumidor (Worker): Un script CLI de PHP que se ejecuta de forma continua en segundo plano, revisando el almacén y procesando los trabajos.

Requisitos previos

• PHP 8.1 o superior.
• Extensiones de PDO y CLI habilitadas.
• Un servidor de base de datos MySQL.

🗄️ Paso 1: Diseño de la Base de Datos para la Cola

Primero, necesitamos una tabla donde almacenar los trabajos pendientes, fallidos o en proceso. Creemos una tabla llamada jobs con los campos necesarios para controlar el estado.

CREATE TABLE `jobs` (
  `id` INT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  `queue` VARCHAR(255) NOT NULL DEFAULT 'default',
  `payload` LONGTEXT NOT NULL,
  `attempts` TINYINT UNSIGNED NOT NULL DEFAULT 0,
  `reserved_at` INT UNSIGNED NULL,
  `available_at` INT UNSIGNED NOT NULL,
  `created_at` INT UNSIGNED NOT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

✍️ Paso 2: Creación de la Estructura de Clases (Jobs y Producer)

Vamos a estructurar nuestro código orientado a objetos. Primero definiremos una interfaz para nuestros trabajos y luego una clase concreta.

La Interfaz del Trabajo

<?php

namespace App\Contracts;

interface JobInterface
{
    public function handle(): void;
}

Un Trabajo Concreto: Envío de Correo

Imaginemos que queremos enviar un correo electrónico de bienvenida de forma asíncrona.

<?php

namespace App\Jobs;

use App\Contracts\JobInterface;

class SendWelcomeEmailJob implements JobInterface
{
    protected string $email;
    protected string $name;

    public function __construct(string $email, string $name)
    {
        $this->email = $email;
        $this->name = $name;
    }

    public function handle(): void
    {
        // Simulación del envío de correo pesado
        echo "Enviando correo a {$this->name} <{$this->email}>...\n";
        sleep(2); // Simulamos retraso de red
        echo "¡Correo enviado exitosamente!\n";
    }
}

El Gestor de Colas (Dispatcher)

Esta clase se encarga de serializar el objeto Job y guardarlo en la base de datos.

<?php

namespace App\Queue;

use App\Contracts\JobInterface;
use PDO;

class QueueDispatcher
{
    protected PDO $db;

    public function __construct(PDO $db)
    {
        $this->db = $db;
    }

    public function push(JobInterface $job, string $queue = 'default'): bool
    {
        $payload = serialize($job);
        $now = time();

        $stmt = $this->db->prepare("INSERT INTO jobs (queue, payload, available_at, created_at) VALUES (?, ?, ?, ?)");
        
        return $stmt->execute([
            $queue,
            $payload,
            $now,
            $now
        ]);
    }
}

⚙️ Paso 3: Construcción del Worker (Consumidor CLI)

El Worker es el corazón del sistema. Es un script diseñado para ejecutarse perpetuamente en la terminal o mediante un administrador de procesos como Supervisor.

<?php

require __DIR__ . '/vendor/autoload.php';

use App\Contracts\JobInterface;

// Configuración de conexión a BD
$db = new PDO('mysql:host=localhost;dbname=queue_db', 'root', 'password', [
    PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION
]);

$queueName = $argv[1] ?? 'default';
echo "Iniciando Queue Worker para la cola: [{$queueName}]...\n";

while (true) {
    // 1. Buscar y reservar el siguiente trabajo de forma atómica
    $jobRecord = getAndReserveNextJob($db, $queueName);

    if (!$jobRecord) {
        // Si no hay trabajos, esperamos 2 segundos para no saturar la CPU
        sleep(2);
        continue;
    }

    $jobId = $jobRecord['id'];
    $payload = $jobRecord['payload'];

    try {
        // 2. Deserializar el objeto Job
        /** @var JobInterface $job */
        $job = unserialize($payload);

        echo "Procesando Job ID: {$jobId}...\n";
        
        // 3. Ejecutar el trabajo
        $job->handle();

        // 4. Eliminar el trabajo completado de la base de datos
        deleteJob($db, $jobId);
        echo "Job ID: {$jobId} completado con éxito.\n";

    } catch (\Throwable $e) {
        echo "Error al procesar Job ID {$jobId}: " . $e->getMessage() . "\n";
        handleFailedJob($db, $jobRecord, $e);
    }
}

function getAndReserveNextJob(PDO $db, string $queue): ?array {
    $db->beginTransaction();

    try {
        $stmt = $db->prepare("SELECT * FROM jobs WHERE queue = ? AND reserved_at IS NULL AND available_at <= ? ORDER BY id ASC LIMIT 1 FOR UPDATE");
        $stmt->execute([$queue, time()]);
        $job = $stmt->fetch(PDO::FETCH_ASSOC);

        if (!$job) {
            $db->rollBack();
            return null;
        }

        // Marcar como reservado incrementando los intentos
        $update = $db->prepare("UPDATE jobs SET reserved_at = ?, attempts = attempts + 1 WHERE id = ?");
        $update->execute([time(), $job['id']]);

        $db->commit();
        return $job;
    } catch (\Exception $e) {
        $db->rollBack();
        throw $e;
    }
}

function deleteJob(PDO $db, int $id): void {
    $stmt = $db->prepare("DELETE FROM jobs WHERE id = ?");
    $stmt->execute([$id]);
}

function handleFailedJob(PDO $db, array $job, \Throwable $e): void {
    // Aquí puedes mover el trabajo a una tabla 'failed_jobs' o reintentarlo
    $stmt = $db->prepare("DELETE FROM jobs WHERE id = ?");
    $stmt->execute([$job['id']]);
}

🧪 Paso 4: Probando el Sistema en Acción

Vamos a simular el flujo completo creando un script productor (dispatch.php) y corriendo nuestro worker.

Script Productor

<?php

require __DIR__ . '/vendor/autoload.php';

use App\Queue\QueueDispatcher;
use App\Jobs\SendWelcomeEmailJob;

$db = new PDO('mysql:host=localhost;dbname=queue_db', 'root', 'password');
$dispatcher = new QueueDispatcher($db);

// Despachar 3 trabajos a la cola
$dispatcher->push(new SendWelcomeEmailJob('ana@example.com', 'Ana Pérez'));
$dispatcher->push(new SendWelcomeEmailJob('carlos@example.com', 'Carlos Gómez'));
$dispatcher->push(new SendWelcomeEmailJob('elena@example.com', 'Elena Torres'));

echo "¡Trabajos despachados con éxito a la cola!\n";

Ejecuta el productor en tu navegador o CLI:

php dispatch.php

Luego, en otra pestaña de tu terminal, enciende el worker:

php worker.php default

Verás en tiempo real cómo el worker procesa secuencialmente cada uno de los correos sin bloquear al usuario.

📊 Comparativa: Ejecución Sincrónica vs. Asíncrona con Colas

🛡️ Buenas Prácticas y Consideraciones de Producción

Construir tu propio sistema en PHP puro es excelente para aprender, pero para llevarlo a producción de forma robusta debes considerar los siguientes puntos:

• Gestión de Procesos con Supervisor: Utiliza una herramienta como Supervisor en entornos Linux para asegurarte de que tu script worker.php se reinicie automáticamente si se cae o si el servidor se reinicia.
• Fugas de Memoria en PHP: Los scripts de larga duración en PHP pueden acumular fugas de memoria (memory leaks). Es recomendable configurar tu worker para que se reinicie automáticamente después de procesar un número determinado de trabajos (por ejemplo, cada 500 jobs).
• Backoff Exponencial: Si un trabajo falla debido a un error temporal de red o base de datos, implementa tiempos de espera incrementales antes de volver a intentarlo.

[program:php-worker]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/html/worker.php default
autostart=true
autorestart=true
user=www-data
numprocs=2
redirect_stderr=true
stdout_log=/var/www/html/storage/worker.log

🎯 Conclusión

Has construido con éxito un sistema completo de gestión de colas y trabajos en segundo plano utilizando exclusivamente PHP puro. Has aprendido sobre serialización de objetos, manejo de transacciones atómicas con bases de datos relacionales, prevención de condiciones de carrera y diseño de arquitecturas orientadas a eventos asíncronos. Esta sólida base te permitirá escalar tus aplicaciones web al siguiente nivel de rendimiento y eficiencia.