Contexto y Necesidad del Procesamiento Asíncrono

En el desarrollo de aplicaciones web, es común enfrentarse a operaciones por lotes que requieren un tiempo de ejecución considerable. Estos procesos pueden ser iniciados por acciones directas del usuario o desencadenados por eventos del sistema. Si se procesan de manera síncrona, el cliente debe mantener la conexión abierta y esperar a que el servidor finalice la tarea, lo cual degrada severamente la experiencia de usuario, especialmente en operaciones pesadas.

Un enfoque asíncrono resuelve este problema. Al igual que los sistemas de correo electrónico modernos que permiten cerrar el navegador inmediatamente después de programar un envío masivo, o los sistemas de comercio electrónico que procesan notificaciones de pedidos en segundo plano, una cola de tareas desacopla la petición HTTP de la ejecución real del proceso pesado.

Evolución Arquitectónica

Una solución básica implica crear tablas específicas para cada tipo de mensaje o tarea y utilizar scripts cronológicos dedicados para procesarlas. Sin embargo, desde una perspectiva de arquitectura de software, es mucho más eficiente diseñar un sistema de colas genérico y abstracto. Esto permite separar la capa de infraestructura de la cola de la lógica de negocio específica, garantizando escalabilidad, reutilización y un mantenimiento simplificado.

Componentes del Sistema

• API de Cola: Interfaces para insertar y extraer tareas.
• Capa de Persistencia: Base de datos relacional para almacenar el estado y los datos de las tareas.
• Worker/Despachador: Proceso en segundo plano (invocado por Cron o como un demonio) que consume y ejecuta las tareas.
• Extensiones: Soporte para prioridades, reintentos y registro de auditoría.

Modelo de Datos

Para almacenar las tareas, se requiere una tabla optimizada para lecturas secuenciales y actualizaciones de estado. A continuación, se presenta una estructura modernizada utilizando utf8mb4 e índices compuestos para mejorar el rendimiento en la extracción de tareas pendientes.

CREATE TABLE `async_task_queue` (
    `id` INT UNSIGNED NOT NULL AUTO_INCREMENT,
    `handler_script` VARCHAR(255) NOT NULL COMMENT 'Ruta o identificador del script ejecutor',
    `payload` TEXT NOT NULL COMMENT 'Datos serializados para el script',
    `is_processed` TINYINT(1) NOT NULL DEFAULT 0,
    `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    `processed_at` TIMESTAMP NULL DEFAULT NULL,
    PRIMARY KEY (`id`),
    INDEX `idx_status_created` (`is_processed`, `created_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

Implementación del Núcleo en PHP

La clase gestora de la cola encapsula la lógica de inserción, lectura y actualización. En lugar de utilizar funciones obsoletas y serialización manual basada en URL, esta implementación emplea PDO para seguridad (prevención de inyección SQL) y json_encode con banderas modernas para una serialización de datos robusta y nativa.

<?php

require_once 'DatabaseConnection.php';

class AsyncTaskQueue
{
    /**
     * Inserta una nueva tarea en la cola.
     *
     * @param string $handler Identificador del script o clase a ejecutar.
     * @param array $payload Datos necesarios para la ejecución.
     * @return int ID de la tarea insertada.
     */
    public function enqueue(string $handler, array $payload): int
    {
        $serializedPayload = json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR);
        
        $sql = "INSERT INTO async_task_queue (handler_script, payload) VALUES (:handler, :payload)";
        $stmt = DatabaseConnection::getInstance()->prepare($sql);
        
        $stmt->execute([
            ':handler' => $handler,
            ':payload' => $serializedPayload
        ]);
        
        return (int) DatabaseConnection::getInstance()->lastInsertId();
    }

    /**
     * Recupera un lote de tareas pendientes.
     *
     * @param int $limit Cantidad máxima de tareas a recuperar.
     * @return array Lista de tareas.
     */
    public function fetchPendingTasks(int $limit = 100): array
    {
        $sql = "SELECT id, handler_script, payload 
                FROM async_task_queue 
                WHERE is_processed = 0 
                ORDER BY created_at ASC 
                LIMIT :limit";
                
        $stmt = DatabaseConnection::getInstance()->prepare($sql);
        $stmt->bindValue(':limit', $limit, PDO::PARAM_INT);
        $stmt->execute();
        
        return $stmt->fetchAll(PDO::FETCH_ASSOC);
    }

    /**
     * Marca una tarea como completada.
     *
     * @param int $taskId ID de la tarea.
     * @return bool Éxito de la operación.
     */
    public function markAsProcessed(int $taskId): bool
    {
        $sql = "UPDATE async_task_queue 
                SET is_processed = 1, processed_at = NOW() 
                WHERE id = :id";
                
        $stmt = DatabaseConnection::getInstance()->prepare($sql);
        return $stmt->execute([':id' => $taskId]);
    }

    /**
     * Deserializa la carga útil de la tarea.
     *
     * @param string $payload Cadena JSON.
     * @return array Datos originales.
     */
    public static function decodePayload(string $payload): array
    {
        try {
            $decoded = json_decode($payload, true, 512, JSON_THROW_ON_ERROR);
            return is_array($decoded) ? $decoded : [];
        } catch (\JsonException $e) {
            return [];
        }
    }
}

Estrategia de Despacho y Ejecución

Para procesar la cola, se implementa un script trabajador (worker) que se ejecuta periódicamente mediante el programador de tareas del sistema operativo (como crontab en Linux) o como un proceso demonio continuo. Este script invoca al método fetchPendingTasks, itera sobre los resultados, decodifica el payload mediante decodePayload, incluye o instancia el handler_script correspondiente, y finalmente llama a markAsProcessed una vez que la lógica de negocio ha finalizado con éxito. Este patrón garantiza que la capa de transporte de la cola permanezca completamente agnóstica respecto a las reglas de negocio específicas que ejecuta.