Driving Message Queues from PHP: RabbitMQ and Redis Streams

1. Why PHP Needs Message Queues

A classic PHP request runs synchronously: the request arrives, PHP-FPM processes it, and the response goes back once everything is done. But as soon as part of that work takes minutes, such as sending an invoice, converting a video, or syncing with a slow third-party system, the request-response cycle becomes a problem. Message queues solve this by moving the actual work out of the request and writing it into a queue that a separate worker process handles at its own pace.

The second major benefit of message queues is decoupling between systems. Instead of a PHP monolith calling an external payment provider directly and becoming unstable itself when that provider fails, it writes a message into a queue. A worker processes that message once the payment provider is reachable again. If the provider is down for ten minutes, messages pile up in the queue instead of customer requests failing with timeout errors.

In PHP projects, two technologies dominate for message queues: RabbitMQ as a dedicated message broker using the AMQP protocol, and Redis Streams as a lightweight alternative that is often already in use for caching. Both approaches are covered in detail in this article, including the pitfalls that show up most often in production PHP applications.

2. RabbitMQ Basics: Exchange, Queue, Binding

RabbitMQ organizes message queues around three central concepts: the exchange, the queue, and the binding between them. A message is never written directly into a queue, but always sent to an exchange first. The exchange decides, based on a routing key and the configured bindings, which queues receive the message. This indirection allows flexible distribution patterns, such as sending one message to multiple queues at once, without the publisher needing to know the recipients.

The direct exchange routes messages based on an exact routing key match, the topic exchange allows wildcard patterns like order.*.created, and the fanout exchange sends every message to all bound queues regardless of routing key. For most PHP applications using message queues, a direct exchange with clearly named routing keys such as order.created or invoice.generate is entirely sufficient and stays easy to follow.

3. Publishing Messages with php-amqplib

The library php-amqplib/php-amqplib is the de facto standard for talking to RabbitMQ from PHP, since it implements the AMQP 0-9-1 protocol entirely in PHP without a native extension. Publishing a message into message queues first means opening a connection and a channel, then declaring the exchange and queue, and only after that sending the actual message. Important: declarations are idempotent, calling queue_declare() repeatedly with identical parameters causes no error, it merely confirms the existing configuration.

For production use, every published message should be marked as persistent, so RabbitMQ writes it to disk and a broker restart does not lose messages. In addition, publisher confirms is recommended, a mechanism where RabbitMQ explicitly confirms to the publisher that the message safely reached the broker before the PHP application assumes delivery succeeded.

<?php

declare(strict_types=1);

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

final class OrderEventPublisher
{
    private AMQPStreamConnection $connection;

    /**
     * Opens a persistent AMQP connection and enables publisher confirms.
     */
    public function __construct(string $host, int $port, string $user, string $password)
    {
        $this->connection = new AMQPStreamConnection($host, $port, $user, $password);
    }

    public function publishOrderCreated(int $orderId, array $payload): void
    {
        $channel = $this->connection->channel();
        $channel->exchange_declare('orders', 'direct', false, true, false);
        $channel->queue_declare('order.created', false, true, false, false);
        $channel->queue_bind('order.created', 'orders', 'order.created');

        // Confirm mode: broker acknowledges receipt before we consider it safe.
        $channel->confirm_select();

        $message = new AMQPMessage(
            json_encode(['order_id' => $orderId] + $payload, JSON_THROW_ON_ERROR),
            [
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
                'content_type' => 'application/json',
            ]
        );

        $channel->basic_publish($message, 'orders', 'order.created');
        $channel->wait_for_pending_acks(5.0);

        $channel->close();
    }
}

4. Writing a Reliable Consumer

A consumer for message queues registers a callback with RabbitMQ and then runs in an infinite loop waiting for incoming messages. The key difference from a classic PHP request is that this process runs permanently, typically as its own systemd service or Supervisor worker, not as part of PHP-FPM. That brings its own operational challenges: memory leaks over hours of runtime, dropped database connections, and the need to restart the process in a controlled way during deployment instead of killing it mid-processing of a message.

Inside the callback, every failure case must be handled explicitly. An unhandled exception in the consumer callback must never silently cause the message to be lost. Instead, the consumer has to decide whether the message should be redelivered (nack with requeue), discarded, or moved to a separate error queue.

<?php

declare(strict_types=1);

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPStreamConnection('rabbitmq.internal', 5672, 'app', 'secret');
$channel = $connection->channel();

$channel->queue_declare('order.created', false, true, false, false);

// Prefetch limits how many unacknowledged messages this worker holds at once.
$channel->basic_qos(null, 10, null);

$callback = function (AMQPMessage $message) use ($channel): void {
    try {
        $payload = json_decode($message->getBody(), true, 512, JSON_THROW_ON_ERROR);
        processOrder($payload);

        // Positive acknowledgment: message is safely removed from the queue.
        $message->ack();
    } catch (\JsonException $e) {
        // Malformed payload will never succeed on retry: discard, no requeue.
        $message->nack(false);
    } catch (\Throwable $e) {
        error_log('Order processing failed: ' . $e->getMessage());
        // Transient failure: requeue for another attempt.
        $message->nack(true);
    }
};

$channel->basic_consume('order.created', '', false, false, false, false, $callback);

while ($channel->is_consuming()) {
    $channel->wait();
}

5. Configuring Acknowledgments and Prefetch Correctly

The acknowledgment model is the heart of reliable message queues. If a consumer does not explicitly confirm a message with ack(), RabbitMQ still considers it unprocessed and redelivers it if the consumer's connection drops. This protects against data loss when a worker process crashes mid-processing, but also creates the need for idempotent processing logic, since the same message can arrive more than once in a failure scenario.

The prefetch value, set via basic_qos(), limits how many unacknowledged messages a consumer may hold at once. A prefetch of 1 processes messages strictly sequentially and is safe but slow. Too high a prefetch value can leave hundreds of messages unacknowledged at once if a worker crashes, all of which then get redelivered. For most message queue use cases in PHP, a prefetch between 5 and 20 is a good starting point, depending on the average processing time per message.

6. Retry Strategies and Dead Letter Queues

Not every error in message queues should be treated the same. A network timeout calling an external service justifies a retry, but a malformed JSON payload will not suddenly become valid on the tenth attempt. A robust retry strategy therefore distinguishes between transient and permanent failures and limits the number of retries, to prevent a broken message from bouncing endlessly between queue and consumer.

RabbitMQ provides x-dead-letter-exchange as a queue argument for this: once a message exceeds the maximum number of retries, or its TTL expires, RabbitMQ automatically moves it into a configured dead letter queue. There, broken messages can be manually inspected, fixed and replayed, instead of being silently lost or blocking the main queue. An additional header such as x-retry-count, maintained by the consumer itself and incremented on every republish, makes the number of attempts so far visible to application logic.

<?php

declare(strict_types=1);

use PhpAmqpLib\Message\AMQPMessage;

// Queue arguments: route exhausted messages to a dead letter exchange.
$arguments = new \PhpAmqpLib\Wire\AMQPTable([
    'x-dead-letter-exchange' => 'orders.dlx',
    'x-dead-letter-routing-key' => 'order.failed',
    'x-message-ttl' => 60000, // 60s before expiry if unprocessed
]);

$channel->queue_declare('order.created', false, true, false, false, false, $arguments);

function republishWithRetryCount(AMQPMessage $message, \PhpAmqpLib\Channel\AMQPChannel $channel, int $maxRetries = 3): void
{
    $headers = $message->get('application_headers');
    $retryCount = $headers !== null && $headers->hasKey('x-retry-count')
        ? (int) $headers->getNativeData()['x-retry-count'] + 1
        : 1;

    if ($retryCount > $maxRetries) {
        $message->nack(false); // exceeded retries, let dead-lettering take over
        return;
    }

    $retryMessage = new AMQPMessage($message->getBody(), [
        'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
        'application_headers' => new \PhpAmqpLib\Wire\AMQPTable(['x-retry-count' => $retryCount]),
    ]);

    $channel->basic_publish($retryMessage, 'orders', 'order.created');
    $message->ack(); // remove original, replaced by the retry copy
}

7. Redis Streams as a Lightweight Alternative

Anyone already running Redis for caching or sessions can use Redis Streams as a much lighter alternative to RabbitMQ for message queues, without operating an extra broker. A Redis stream is an append-only log structure into which entries are written with XADD, and from which consumer groups read messages with XREADGROUP, similar to the Kafka model but without its operational complexity.

Consumer groups in Redis Streams offer their own acknowledgment model via XACK, functionally comparable to RabbitMQ's ack mechanism. Unacknowledged entries remain visible in the pending entries list and can be claimed by another consumer with XCLAIM if the original worker went down. The big advantage over RabbitMQ is operational simplicity, the drawback is less sophisticated routing, there are no exchange concepts, only named streams.

<?php

declare(strict_types=1);

$redis = new \Redis();
$redis->connect('redis.internal', 6379);

// Publisher: append an entry to the stream.
$redis->xAdd('orders:created', '*', [
    'order_id' => '10245',
    'total' => '129.90',
]);

// Consumer group setup, run once during deployment.
try {
    $redis->xGroup('CREATE', 'orders:created', 'invoice-workers', '0');
} catch (\RedisException $e) {
    // Group already exists — safe to ignore on redeploy.
}

// Worker loop: read new entries assigned to this consumer group.
while (true) {
    $entries = $redis->xReadGroup(
        'invoice-workers',
        'worker-1',
        ['orders:created' => '>'],
        10,
        5000
    );

    foreach ($entries['orders:created'] ?? [] as $id => $fields) {
        try {
            generateInvoice($fields);
            $redis->xAck('orders:created', 'invoice-workers', $id);
        } catch (\Throwable $e) {
            error_log('Invoice generation failed for ' . $id . ': ' . $e->getMessage());
            // Left unacknowledged — visible in pending entries for retry via XCLAIM.
        }
    }
}

8. Idempotency: Safely Preventing Duplicate Processing

Both RabbitMQ and Redis Streams guarantee "at least once" delivery by default, never "exactly once". This means every message can theoretically arrive at the consumer more than once, for example when the confirmation gets lost after successful processing but before the ack is sent. Anyone using message queues in production must therefore make every piece of processing logic idempotent, so duplicate delivery does not cause duplicate effects.

The most common solution is a unique idempotency key per message, usually a UUID, checked against a table or a Redis set before the actual processing happens. If the key already exists, the message is marked as already processed and skipped, without executing the business logic again. For operations like sending an invoice email or charging a payment amount, this safeguard is not optional but a necessary prerequisite for using message queues safely in payment processes at all.

9. RabbitMQ vs. Redis Streams Compared

Choosing between RabbitMQ and Redis Streams for message queues in PHP projects depends heavily on existing operational constraints and routing flexibility requirements. The table below summarizes the key differences.

For complex distribution patterns with multiple independent recipients and strict delivery guarantees, RabbitMQ remains the more robust choice among message queue technologies. For smaller PHP projects already relying on Redis and needing simple worker queues, Redis Streams save operational complexity without sacrificing fundamental reliability features like consumer groups and acknowledgments.

10. Summary

Message queues solve a central problem of synchronous PHP applications: slow or unreliable work no longer has to block the request-response cycle. RabbitMQ with php-amqplib offers mature routing through exchanges, reliable acknowledgments and native dead letter queues. Redis Streams win on lower operational overhead when Redis is already part of the stack, at the cost of slightly simpler routing logic.

Regardless of the chosen technology, the same basic rules apply: store messages persistently, limit prefetch, distinguish transient from permanent failures, and make every processing step idempotent. Anyone who consistently applies these principles when using message queues builds PHP systems that stay stable and traceable even when third-party systems fail.

11. FAQ: Driving Message Queues from PHP