Publish/Subscribe

In this tutorial, we will use PHP with the php-amqplib library to implement a publish/subscribe system. This pattern differs from work queues, where each task is delivered to exactly one worker. Instead, we will broadcast messages to multiple consumers.

Prerequisites

This tutorial assumes RabbitMQ is installed and running on localhost at the standard port (5672). If you use a different host, port, or credentials, adjust the connection settings accordingly.

Help and Support

If you ancounter difficulties while reading this tutorial, you can seek assistance via the mailing list.

Exchanges

In previous tutorials, we covered producers sending messages to queues and consumers receiving them. Now, let's delve into RabbitMQ's full messaging model.

• A producer is an application that sends messages.
• A queue is a buffer that stores messages.
• A consumer is an application that receives messages.

Producers never send messages directly to queues. Instead, they send messages to an exchange. An exchange determines how to route messages to queues based on rules defined by its type (direct, topic, headers, fanout).

We'll focus on the fanout exchange type, which broadcasts all received messages to all known queues. First, declare a fanout exchange named logs:

$channel->exchange_declare('logs', 'fanout', false, false, false);

Temporary Queues

For our logging system, we need each consumer to recieve all log messages. We also want to process only current messages, ignoring old ones. To achieve this, we need:

• A unique queue for each consumer instance.
• The queue to be deleted when the consumer disconnects.

Create a non-durable, exclusive queue with a server-generated name:

list($queue_name, ,) = $channel->queue_declare("");

Binding

To ensure messages from the logs exchange are routed to our queue, bind the queue to the exchange:

$channel->queue_bind($queue_name, 'logs');

Complete Code

The producer script, emit_logs.php, publishes messages to the logs exchange:

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

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

$channel->exchange_declare('logs', 'fanout', false, false, false);

$data = implode(' ', array_slice($argv, 1));
if (empty($data)) {
    $data = "info: Hello World!";
}
$msg = new AMQPMessage($data);

$channel->basic_publish($msg, 'logs');

echo ' [x] Sent ', $data, "\n";

$channel->close();
$connection->close();
?>

The consumer script, receive_logs.php, listans for messages from the logs exchange:

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

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

$channel->exchange_declare('logs', 'fanout', false, false, false);

list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);

$channel->queue_bind($queue_name, 'logs');

echo " [*] Waiting for logs. To exit press CTRL+C\n";

$callback = function ($msg) {
    echo ' [x] ', $msg->body, "\n";
};

$channel->basic_consume($queue_name, '', false, true, false, false, $callback);

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

$channel->close();
$connection->close();
?>

To save logs to a file, run:

php receive_logs.php > logs_from_rabbit.log

To view logs on the screen, open another terminal and execute:

php receive_logs.php

Emit logs by running:

php emit_logs.php

Use rabbitmqctl list_bindings to verify bindings and queues. After running two instances of receive_logs.php, you should see output similar to:

sudo rabbitmqctl list_bindings
# => Listing bindings ...
# => logs    exchange        amq.gen-JzTY20BRgKO-HjmUJj0wLg  queue           []
# => logs    exchange        amq.gen-vso0PVvyiRIL2WoV3i48Yg  queue           []
# => ...done.