Publish/Subscribe
Publish/Subscribe
(using php-amqplib)
Prerequisites
This tutorial assumes RabbitMQ is installed and running
on localhost on standard port (5672). In case you use
a different host, port or credentials, connections settings would require adjusting.
Where to get help
If you're having trouble going through this tutorial you can
through the mailing list.
In the previous tutorial we created a work
queue. The assumption behind a work queue is that each task is
delivered to exactly one worker. In this part we'll do something
completely different -- we'll deliver a message to multiple
consumers. This pattern is known as "publish/subscribe".
To illustrate the pattern, we're going to build a simple logging
system. It will consist of two programs -- the first will emit log
messages and the second will receive and print them.
In our logging system every running copy of the receiver program will
get the messages. That way we'll be able to run one receiver and
direct the logs to disk; and at the same time we'll be able to run
another receiver and see the logs on the screen.
Essentially, published log messages are going to be broadcast to all
the receivers.
Exchanges
In previous parts of the tutorial we sent and received messages to and
from a queue. Now it's time to introduce the full messaging model in
Rabbit.
Let's quickly go over what we covered in the previous tutorials:
The core idea in the messaging model in RabbitMQ is that the producer
never sends any messages directly to a queue. Actually, quite often
the producer doesn't even know if a message will be delivered to any
queue at all.
Instead, the producer can only send messages to an exchange. An
exchange is a very simple thing. On one side it receives messages from
producers and the other side it pushes them to queues. The exchange
must know exactly what to do with a message it receives. Should it be
appended to a particular queue? Should it be appended to many queues?
Or should it get discarded. The rules for that are defined by the
exchange type.
digraph {
bgcolor=transparent;
truecolor=true;
rankdir=LR;
node [style="filled"];
//
P1 [label="P", fillcolor="#00ffff"];
X [label="X", fillcolor="#3333CC"];
Q1 [label="{||||}", fillcolor="red", shape="record"];
Q2 [label="{||||}", fillcolor="red", shape="record"];
//
P1 -> X;
X -> Q1;
X -> Q2;
}
There are a few exchange types available: direct, topic, headers
and fanout. We'll focus on the last one -- the fanout. Let's create
an exchange of this type, and call it logs:
$channel->exchange_declare('logs', 'fanout', false, false, false);
The fanout exchange is very simple. As you can probably guess from the
name, it just broadcasts all the messages it receives to all the
queues it knows. And that's exactly what we need for our logger.
Listing exchanges
To list the exchanges on the server you can run the ever useful rabbitmqctl:
sudo rabbitmqctl list_exchanges
In this list there will be some amq.* exchanges and the default (unnamed)
exchange. These are created by default, but it is unlikely you'll need to
use them at the moment.
The default exchange
In previous parts of the tutorial we knew nothing about exchanges,
but still were able to send messages to queues. That was possible
because we were using a default exchange, which we identify by the empty string ("").
Recall how we published a message before:
$channel->basic_publish($msg, '', 'hello');
Here we use the default or nameless exchange: messages are
routed to the queue with the name specified by routing_key, if it exists.
The routing key is the third argument to basic_publish
Now, we can publish to our named exchange instead:
$channel->exchange_declare('logs', 'fanout', false, false, false);
$channel->basic_publish($msg, 'logs');
Temporary queues
As you may remember previously we were using queues that had
specific names (remember hello and task_queue?). Being able to name
a queue was crucial for us -- we needed to point the workers to the
same queue. Giving a queue a name is important when you
want to share the queue between producers and consumers.
But that's not the case for our logger. We want to hear about all
log messages, not just a subset of them. We're
also interested only in currently flowing messages not in the old
ones. To solve that we need two things.
Firstly, whenever we connect to Rabbit we need a fresh, empty queue.
To do this we could create a queue with a random name, or,
even better - let the server choose a random queue name for us.
Secondly, once we disconnect the consumer the queue should be
automatically deleted.
In the php-amqplib client, when we supply queue name
as an empty string, we create a non-durable queue with a generated name:
list($queue_name, ,) = $channel->queue_declare("");
When the method returns, the $queue_name variable contains a random queue name
generated by RabbitMQ. For example it may look like amq.gen-JzTY20BRgKO-HjmUJj0wLg.
When the connection that declared it closes, the queue will be deleted
because it is declared as exclusive. You can learn more about the exclusive flag and other queue
properties in the guide on queues.
Bindings
digraph {
bgcolor=transparent;
truecolor=true;
rankdir=LR;
node [style="filled"];
//
P1 [label="P", fillcolor="#00ffff"];
X [label="X", fillcolor="#3333CC"];
Q1 [label="{||||}", fillcolor="red", shape="record"];
Q2 [label="{||||}", fillcolor="red", shape="record"];
//
P1 -> X;
X -> Q1 [label="binding"];
X -> Q2 [label="binding"];
}
We've already created a fanout exchange and a queue. Now we need to
tell the exchange to send messages to our queue. That relationship
between exchange and a queue is called a binding.
$channel->queue_bind($queue_name, 'logs');
From now on the logs exchange will append messages to our queue.
Listing bindings
You can list existing bindings using, you guessed it,
rabbitmqctl list_bindings
Putting it all together
digraph {
bgcolor=transparent;
truecolor=true;
rankdir=LR;
node [style="filled"];
//
P [label="P", fillcolor="#00ffff"];
X [label="X", fillcolor="#3333CC"];
subgraph cluster_Q1 {
label="amq.gen-RQ6...";
color=transparent;
Q1 [label="{||||}", fillcolor="red", shape="record"];
};
subgraph cluster_Q2 {
label="amq.gen-As8...";
color=transparent;
Q2 [label="{||||}", fillcolor="red", shape="record"];
};
C1 [label=<C<font point-size="7">1</font>>, fillcolor="#33ccff"];
C2 [label=<C<font point-size="7">2</font>>, fillcolor="#33ccff"];
//
P -> X;
X -> Q1;
X -> Q2;
Q1 -> C1;
Q2 -> C2;
}
The producer program, which emits log messages, doesn't look much
different from the previous tutorial. The most important change is that
we now want to publish messages to our logs exchange instead of the
nameless one. Here goes the code for
emit_log.php script:
<?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();
(emit_log.php source)
As you see, after establishing the connection we declared the
exchange. This step is necessary as publishing to a non-existing
exchange is forbidden.
The messages will be lost if no queue is bound to the exchange yet,
but that's okay for us; if no consumer is listening yet we can safely discard the message.
The code for receive_logs.php:
<?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 (count($channel->callbacks)) {
$channel->wait();
}
$channel->close();
$connection->close();
(receive_logs.php source)
If you want to save logs to a file, just open a console and type:
php receive_logs.php > logs_from_rabbit.log
If you wish to see the logs on your screen, spawn a new terminal and run:
php receive_logs.php
And of course, to emit logs type:
php emit_log.php
Using rabbitmqctl list_bindings you can verify that the code actually
creates bindings and queues as we want. With two receive_logs.php
programs running you should see something like:
sudo rabbitmqctl list_bindings
# => Listing bindings ...
# => logs exchange amq.gen-JzTY20BRgKO-HjmUJj0wLg queue []
# => logs exchange amq.gen-vso0PVvyiRIL2WoV3i48Yg queue []
# => ...done.
The interpretation of the result is straightforward: data from
exchange logs goes to two queues with server-assigned names. And
that's exactly what we intended.
To find out how to listen for a subset of messages, let's move on to
tutorial 4
Production [Non-]Suitability Disclaimer
Please keep in mind that this and other tutorials are, well, tutorials.
They demonstrate one new concept at a time and may intentionally oversimplify some things and leave out others. For example topics such as
connection management, error handling, connection recovery, concurrency and metric collection are largely omitted
for the sake of brevity. Such simplified code should not be considered production ready.
Please take a look at the rest of the documentation before going live with your app.
We particularly recommend the following guides: Publisher Confirms and Consumer Acknowledgements,
Production Checklist and Monitoring.
Getting Help and Providing Feedback
If you have questions about the contents of this tutorial or
any other topic related to RabbitMQ, don't hesitate to ask them
on the .
Help Us Improve the Docs <3
If you'd like to contribute an improvement to the site,
its source is available on GitHub.
Simply fork the repository and submit a pull request. Thank you!
