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.

Nameless 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.basicPublish("","hello",null, message.getBytes());

The first parameter is the name of the exchange. The empty string denotes the default or nameless exchange: messages are routed to the queue with the name specified by routingKey, if it exists.

Now, we can publish to our named exchange instead:

channel.basicPublish("logs","",null, message.getBytes());

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 Java client, when we supply no parameters to queueDeclare() we create a non-durable, exclusive, autodelete queue with a generated name:

String queueName = channel.queueDeclare().getQueue();

You can learn more about the exclusive flag and other queue properties in the guide on queues.

At that point queueName contains a random queue name. For example it may look like amq.gen-JzTY20BRgKO-HjmUJj0wLg.

Bindings​

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.queueBind(queueName,"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​

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. We need to supply a routingKey when sending, but its value is ignored for fanout exchanges. Here goes the code for EmitLog.java program:

publicclassEmitLog{

privatestaticfinalStringEXCHANGE_NAME="logs";

publicstaticvoidmain(String[] argv)throwsException{
ConnectionFactory factory =newConnectionFactory();
 factory.setHost("localhost");
try(Connection connection = factory.newConnection();
Channel channel = connection.createChannel()){
 channel.exchangeDeclare(EXCHANGE_NAME,"fanout");

String message = argv.length <1?"info: Hello World!":
String.join(" ", argv);

 channel.basicPublish(EXCHANGE_NAME,"",null, message.getBytes("UTF-8"));
System.out.println(" [x] Sent '"+ message +"'");
}
}
}

(EmitLog.java 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 ReceiveLogs.java:

importcom.rabbitmq.client.Channel;
importcom.rabbitmq.client.Connection;
importcom.rabbitmq.client.ConnectionFactory;
importcom.rabbitmq.client.DeliverCallback;

publicclassReceiveLogs{
privatestaticfinalStringEXCHANGE_NAME="logs";

publicstaticvoidmain(String[] argv)throwsException{
ConnectionFactory factory =newConnectionFactory();
 factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

 channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
String queueName = channel.queueDeclare().getQueue();
 channel.queueBind(queueName,EXCHANGE_NAME,"");

System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

DeliverCallback deliverCallback =(consumerTag, delivery)->{
String message =newString(delivery.getBody(),"UTF-8");
System.out.println(" [x] Received '"+ message +"'");
};
 channel.basicConsume(queueName,true, deliverCallback, consumerTag ->{});
}
}

(ReceiveLogs.java source)

Compile as before and we're done.

javac -cp$CP EmitLog.java ReceiveLogs.java

If you want to save logs to a file, just open a console and type:

java-cp$CP ReceiveLogs > logs_from_rabbit.log

If you wish to see the logs on your screen, spawn a new terminal and run:

java-cp$CP ReceiveLogs

And of course, to emit logs type:

java-cp$CP EmitLog

Using rabbitmqctl list_bindings you can verify that the code actually creates bindings and queues as we want. With two ReceiveLogs.java 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