PHP Academy · Lección

Trabajo con RabbitMQ en PHP

Publique y consuma mensajes con RabbitMQ.

Trabajo con RabbitMQ en PHP es una lección gratuita de PHP Academy en CoddyKit. Esta es la lección 2 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de PHP Academy, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de PHP Academy incluye 4 lecciones en total.

RabbitMQ en PHP

RabbitMQ es un broker que utiliza AMQP 0-9-1. En PHP, el cliente de referencia es php-amqplib/php-amqplib (PHP puro) o la extensión respaldada por C ext-amqp. Esta lección utiliza php-amqplib porque se ejecuta en cualquier entorno donde funcione Composer.

El modelo AMQP básico tiene tres actores: los exchanges reciben mensajes, los bindings los enrutan mediante una clave y las queues los almacenan para los consumidores. Domine este modelo y el resto serán detalles.

Instalación del cliente

Añada la biblioteca con Composer. Necesita las extensiones sockets y bcmath, habituales ambas en las compilaciones de PHP para CLI.

composer require php-amqplib/php-amqplib
# Connection target, e.g. amqp://guest:guest@localhost:5672/

El modelo de exchange, cola y binding

Los productores publican en un exchange, nunca directamente en una cola. El tipo de exchange determina el enrutamiento:

direct: coincidencia exacta con la clave de enrutamiento.
topic: patrones con comodines como order.*.eu.
fanout: difusión a todas las colas vinculadas.
headers: coincidencia basada en los atributos de las cabeceras.

Un binding conecta una cola con un exchange mediante un patrón de enrutamiento. Desacoplar los productores de la topología de las colas es precisamente el objetivo de la capa de exchange.

Conexión y declaraciones

Abra una conexión, obtenga un canal y declare la topología. durable: true hace que el exchange y la cola sobrevivan al reinicio del broker. Las declaraciones son idempotentes: declarar una entidad existente con argumentos coincidentes no produce ningún efecto.

<?php
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();

$ch->exchange_declare('orders', 'topic', false, true, false);
$ch->queue_declare('orders.email', false, true, false, false);
$ch->queue_bind('orders.email', 'orders', 'order.created');

echo "Topology ready\n";
$ch->close();
$conn->close();

Publicar un mensaje

Envuelva el cuerpo en un AMQPMessage. Establezca delivery_mode = 2 para que el mensaje sea persistente: la combinación de una cola durable y un mensaje persistente es la que sobrevive a un reinicio; una sin la otra no lo hace.

<?php
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();

$payload = json_encode(['orderId' => 42, 'total' => 19.90]);
$msg = new AMQPMessage($payload, [
    'content_type'  => 'application/json',
    'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
    'message_id'    => bin2hex(random_bytes(8)),
]);

$ch->basic_publish($msg, 'orders', 'order.created');
echo "Published\n";
$ch->close();
$conn->close();

Consumir con ack manual

Registre un callback con basic_consume. Pase no_ack = false para controlar la confirmación. Envíe el ack solo después de que el trabajo se complete correctamente; si falla, use basic_nack con requeue para que RabbitMQ vuelva a entregar el mensaje.

<?php
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();
$ch->basic_qos(null, 10, null); // prefetch 10

$callback = function ($msg) {
    $data = json_decode($msg->getBody(), true);
    try {
        // ... do work ...
        $msg->ack();
    } catch (\Throwable $e) {
        $msg->nack(true); // requeue
    }
};

$ch->basic_consume('orders.email', '', false, false, false, false, $callback);
while ($ch->is_consuming()) {
    $ch->wait();
}

Prefetch y distribución justa

De forma predeterminada, RabbitMQ distribuye los mensajes entre los consumidores mediante round-robin, sin tener en cuenta lo ocupado que está cada uno; un consumidor lento acumula trabajo. basic_qos(null, prefetch, null) limita el número de mensajes sin confirmar que puede mantener un consumidor.

Establezca prefetch en un número pequeño (por ejemplo, 1–10) para tareas pesadas y desiguales, de modo que el broker envíe trabajo nuevo solo a los consumidores que tengan capacidad disponible. Esto se denomina distribución justa.

Confirmaciones del publicador

basic_publish retorna inmediatamente y no le indica si el broker aceptó el mensaje. Para publicar con garantías, habilite las confirmaciones del publicador: el broker envía un ack cuando el mensaje se ha persistido o enrutado de forma segura.

<?php
require 'vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();
$ch->confirm_select(); // enable confirms on this channel

$ch->set_ack_handler(fn($m) => print("confirmed\n"));
$ch->set_nack_handler(fn($m) => print("REJECTED\n"));

$ch->basic_publish(new AMQPMessage('hi'), 'orders', 'order.created');
$ch->wait_for_pending_acks(5.0); // block until confirmed or timeout

Enrutamiento a mensajes muertos

Configure el argumento x-dead-letter-exchange de una cola para que los mensajes rechazados (nack sin requeue) o caducados se enruten a un DLX. Combine esta configuración con x-delivery-limit en las colas de quórum para limitar automáticamente los reintentos.

<?php
use PhpAmqpLib\Wire\AMQPTable;

$args = new AMQPTable([
    'x-dead-letter-exchange'    => 'orders.dlx',
    'x-dead-letter-routing-key' => 'order.failed',
    'x-message-ttl'             => 60000, // ms before expiry
]);
// false=passive, true=durable, false=exclusive, false=autodelete, args
$ch->queue_declare('orders.email', false, true, false, false, false, $args);
$ch->queue_declare('orders.dead', false, true, false, false);
$ch->queue_bind('orders.dead', 'orders.dlx', 'order.failed');

Ejecutar workers en producción

Estas son algunas reglas operativas para workers de RabbitMQ en PHP, fruto de la experiencia:

Ejecute los consumidores como procesos CLI de larga duración bajo un supervisor (systemd / Supervisor) que los reinicie al terminar.
PHP pierde memoria con el tiempo: reinicie el worker después de N mensajes o al alcanzar un umbral de memoria.
Envíe heartbeats de AMQP y gestione SIGTERM para realizar un apagado ordenado (termine el mensaje en curso y deténgase).
Use colas de quórum para alta disponibilidad en lugar de las colas reflejadas heredadas.

Apagado ordenado

Los despliegues envían SIGTERM. Un worker ingenuo muere a mitad del mensaje, lo que obliga a volver a entregarlo. Instale un controlador de señales que cambie una bandera; termine el mensaje actual, envíe su ack y, después, salga limpiamente del bucle de consumo. pcntl_async_signals(true) permite que PHP entregue la señal entre las esperas de AMQP.

<?php
pcntl_async_signals(true);
$running = true;
pcntl_signal(SIGTERM, function () use (&$running) {
    $running = false; // stop after the current message
    echo "SIGTERM: draining...\n";
});

while ($running && $ch->is_consuming()) {
    try {
        $ch->wait(null, false, 5); // wakes for signals
    } catch (\PhpAmqpLib\Exception\AMQPTimeoutException $e) {
        // idle tick - loop and re-check $running
    }
}
$ch->close();
echo "Stopped cleanly\n";

Comprobación rápida

Sobrevivir al reinicio de un broker.

Resumen

Ahora puede crear un pipeline real de RabbitMQ en PHP:

Publique en un exchange y enrute mediante bindings hacia las colas.
Una cola durable + un mensaje persistente sobreviven a los reinicios.
Consuma con ack manual y ajuste basic_qos prefetch para lograr una distribución justa.
Use confirmaciones del publicador para garantizar los envíos y un DLX para los mensajes problemáticos.
Ejecute los workers bajo un supervisor, con heartbeats y apagado ordenado.

Siguiente tema: Kafka, cuando necesite streaming de alto rendimiento en lugar de una cola de tareas.

Preguntas frecuentes

¿La lección «Trabajo con RabbitMQ en PHP» es gratis?

Sí — el texto completo de «Trabajo con RabbitMQ en PHP» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de PHP Academy, actualiza a CoddyKit PRO. El curso de PHP Academy incluye 4 lecciones en total.

¿Qué aprenderé en «Trabajo con RabbitMQ en PHP»?

Publique y consuma mensajes con RabbitMQ. Practicas PHP Academy con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar PHP Academy?

No se requiere experiencia previa. PHP Academy en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 2 de 4.

¿Cuánto tiempo toma la lección «Trabajo con RabbitMQ en PHP»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de PHP Academy?

Sí. Cada lección de PHP Academy incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

Por qué usar mensajería asíncrona
Trabajo con RabbitMQ en PHP
Apache Kafka con PHP
Creación de flujos de trabajo basados en eventos

← Volver a PHP Academy