RabbitMQ Cookbook

To use this recipe we need to set up the Java development environment as mentioned in the Introduction section.

In order to create a Java client that connects to the RabbitMQ broker, you need to perform the following steps:

Using the Java client API, the application must create an instance of ConnectionFactory and set the host where RabbitMQ should be running with the setHost() method.

After the Java imports (step 1), we have instantiated the factory object (step 2). In this example we have just set the hostname that we have chosen to optionally get from the command line (step 3), but you can find more information regarding connection options in the section There's more….

In step 4 we have actually established the TCP connection to the RabbitMQ broker.

However, we are not yet ready to communicate with the broker; we need to set up a communication channel (step 5). This is an advanced concept of AMQP; using this abstraction, it is possible to let many different messaging sessions use the same logical connection.

Actually, all the communication operations of the Java client library are performed by the methods of a channel instance.

If you are developing multithreaded applications, it is highly recommended to use a different channel for each thread. If many threads use the same channel, they will serialize their execution in the channel method calls, leading to possible performance degradation.

It is possible to specify many different optional properties for any RabbitMQ connection. You can find them all in the online documentation at (http://www.rabbitmq.com/releases/rabbitmq-java-client/current-javadoc). These options are all self-explanatory, except for the AMQP virtual host.

Virtual hosts are administrative containers; they allow to configure many logically independent brokers hosts within one single RabbitMQ instance, to let many different independent applications share the same RabbitMQ server. Each virtual host can be configured with its independent set of permissions, exchanges, and queues and will work in a logically separated environment.

It's possible to specify connection options by using just a connection string, also called connection URI, with the factory.setUri() method:

ConnectionFactory factory = new ConnectionFactory();
String uri="amqp://user:pass@hostname:port/vhost";
factory.setUri(uri);

To use this recipe we need to set up the Java development environment as indicated in the Introduction section.

After connecting to the broker, as seen in the previous recipe, you can start sending messages performing the following steps:

In this first basic example we have been able to just send a message to RabbitMQ.

After the communication channel is established, the first step is to ensure that the destination queue exists. This task is accomplished declaring the queue (step 1) calling queueDeclare(). The method call does nothing if the queue already exists, otherwise it creates the queue itself.

Note that this, as most of the AMQP operations, is a method of the Channel Java interface. All the operations that need interactions with the broker are carried out through channels.

Let's examine the meaning of the queueDeclare() method call in depth. Its template can be found in the Java client reference documentation located at http://www.rabbitmq.com/releases/rabbitmq-java-client/current-javadoc/. The documentation will be as shown in the following screenshot:

In particular we have used the second overload of this method that we report here:

AMQP.Queue.DeclareOk queueDeclare(java.lang.String queue, boolean durable, boolean exclusive, booleanautoDelete, java.util.Map<java.lang.String,java.lang.Object> arguments) throws java.io.IOException

The meanings of the individual arguments are:

In step 2 we have actually sent a message to the RabbitMQ broker.

The message body will never be opened by RabbitMQ. Messages are opaque entities for the AMQP broker, and you can use any serialization format you like. We often use JSON, but XML, ASN.1, standard or custom, ASCII or binary format, are all valid alternatives. The only important thing is that the client applications should know how to interpret the data.

Let's now examine in depth the basicPublish() method of the Channel interface for the overload used in our recipe:

void basicPublish(java.lang.String exchange, java.lang.String routingKey, AMQP.BasicProperties props, byte[] body) throws java.io.IOException

In our example the exchange argument has been set to the empty string "", that is, the default exchange, and the routingKey argument to the name of the queue. In this case the message is directly sent to the queue specified as routingKey. The body argument is set to the byte array of our string, that is, just the message that we sent. The props argument is set to null as a default; these are the message properties, discussed in depth in the recipe Using message properties.

For example, in step 3 we have sent an identical message, but with props set to MessageProperties.PERSISTENT_TEXT_PLAIN; in this way we have requested RabbitMQ to mark this message as a persistent message.

Both the messages have been dispatched to the RabbitMQ broker, logically queued in the myFirstQueue queue. The messages will stay buffered there until a client, (typically, a different client) gets it.

If the queue has been declared with the durable flag set to true and the message has been marked as persistent, it is stored on the disk by the broker. If one of the two conditions is missing, the message is stored in the memory. In the latter case the buffered messages won't survive a RabbitMQ restart, but the message delivery and retrieval will be much faster. However, we will dig down on this topic in Chapter 8, Performance Tuning for RabbitMQ.

In this section we will discuss the methods to check the status of RabbitMQ and whether a queue already exists.

How to check the RabbitMQ status

In order to check the RabbitMQ status, you can use the command-line control tool rabbitmqctl. It should be in the PATH in the Linux setup. On Windows it can be found running the RabbitMQ command shell by navigating to Start Menu | All Programs | RabbitMQ Server | RabbitMQ Command Prompt (sbin dir). We can run rabbitmqctl.bat from this command prompt.

We can check the queue status with the command rabbitmqclt list_queues. In the following screenshot, we have run it just before and after we have run our example.

We can see our myfirstqueue queue listed in the preceding screenshot, followed by the number 2, which is just the number of the messages buffered into our queue.

Now we can either try to restart RabbitMQ, or reboot the machine hosting it. Restarting RabbitMQ successfully will depend on the used OS:

• On Linux, RedHat, Centos, Fedora, Raspbian, and so on:

  service rabbitmq-server restart
  

• On Linux, Ubuntu, Debian, and so on:

  /etc/init.d/rabbitmq restart
  

• On Windows:

  sc stop rabbitmq / sc start rabbitmq
  

How many messages should we expect when we run rabbitmqclt list_queues again?

To use this recipe we need to set up the Java development environment as indicated in the introduction.

In order to consume the messages sent as seen in the previous recipe, perform the following steps:

After we have established the connection and the channel to the AMQP broker as seen in the Connecting to the broker recipe, we need to ensure that the queue from which we are going to consume the messages exists (step 1).

In fact it is possible that the consumer is started before any producer has sent a message to the queue and the queue itself may actually not exist at all. To avoid the failure of the subsequent operations on the queue, we need to declare the queue.

The heart of this recipe is step 2. Here we have defined our specialized consumer that overrides handleDelivery() and instantiated it in step 3. In the Java client API the consumer callbacks are defined by the com.rabbitmq.client.Consumer interface. We have extended our consumer from DefaultConsumer, which provides a no-operation implementation for all the methods declared in the Consumer interface.

In step 3, by calling channel.basicConsume(), we let the consumer threads start consuming messages. The consumers of each channel are always executed on the same thread, independent of the calling one.

Now that we have activated a consumer for myQueue, the Java client library starts getting messages from that queue on the RabbitMQ broker, and invokes handleDelivery() for each one.

Then after the channel.basicConsume() method's invocation, we just sit idle waiting for a key press in the main thread. Messages are being consumed with nonblocking semantics respecting the event-driven paradigm, typical of messaging applications.

Only after we press Enter, the execution proceeds to step 5, cancelling the consumer. At this point the consumer threads stop invoking our consumer object, and we can release the resources and exit.

In this section we will learn more about consumer threads and the use of blockage semantics.

ExecutorService eService = Executors.newFixedThreadPool(10);
Connection connection = factory.newConnection(eService);

eService.shutdown();

You can find all the available methods of the consumer interface in the official Javadoc at

http://www.rabbitmq.com/releases/rabbitmq-java-client/current-javadoc/com/rabbitmq/client/Consumer.html

To use this recipe you will need to set up Java and Python environments as described in the introduction.

To implement a Java producer and a Python consumer, you can perform the following steps:

After we set up the environments (step 1 and step 2), we serialize the newbooks class with the method write(newbooks). The method returns a JSON String (jsonmessage) as shown in the following code snippet:

[
  {
    "author" : "John Doe",
    "bookDescription" : "History VOL: 1",
    "bookID" : 1
  },
  {
    "author" : "John Doe",
    "bookDescription" : "History VOL: 2",
    "bookID" : 2
  }
]

In step 4 we publish jsonmessage to the queue myJSONBodyQueue_4. Now the Python Consumer can get the message from the same queue. Let's see how to do it in Python:

  connection = pika.BlockingConnection(pika.ConnectionParameters(rabbitmq_host));
  channel = connection.channel()
  queue_name = "myJSONBodyQueue_4"
  channel.queue_declare(queue=my_queue)
  ..
  channel.basic_consume(consumer_callback, queue=my_queue, no_ack=True) 
  channel.start_consuming()

As we have seen in the Java implementation, we must create a connection and then create a channel. With the method channel.queue_declare(queue=myQueue), we declare a queue that is not durable, exclusive or autodelete. In order to change the queue's attribute, it's enough to add the parameter in the queue_declare method as follows:

channel.queue_declare(queue=myQueue,durable=True)

With the method channel.basic_consume(), the client starts consuming messages from the given queue, invoking the callback consumer_callback()where it will receive the messages.

While the callbacks in Java were defined in the consumer interface, in Python they are just passed to basic_consume(), in spite of the more functional, less declarative, and less formal paradigm typical of Python.

The callback consumer_callback is as follows:

def consumer_callback(ch, method, properties, body):
  newBooks=json.loads(body); 
  print" Count books:",len(newBooks);
  for item in newBooks:
    print 'ID:',item['bookID'], '-Description:',item['bookDescription'],' -Author:',item['author']

The callback takes the message, deserializes it with json.loads(), and then the newBooks structure is ready to be read.

The JSON helper tools included in the RabbitMQ client library are very simple, but in a real project you can evaluate them to use an external JSON library. For example, a powerful Java JSON library is google-gson (https://code.google.com/p/google-gson/) or jackson (http://jackson.codehaus.org/).

To use this recipe we need to set up the Java development environment as indicated in the Introduction section.

Let's perform the following steps to implement the RPC responder:

Now let's perform the following steps to implement the RPC caller:

In this example the RPC responder takes the role of an RPC server; the responder listens on the requestQueue public queue (step 1), where the callers will place their requests.

Each caller, on the other hand, will consume the responder replies on its own private queue, created in step 5.

When the caller sends a message (step 11), it includes two properties: the name of the temporary reply queue (replyTo()) where it will be listening, and a message identifier (correlationId()), needed by the caller to identify the call when the reply comes back.

In fact, in our example we have implemented an asynchronous RPC caller. The action to be performed by the RpcCallerConsumer (step 6) when the reply comes back is recorded by the nonblocking consumer by calling AddAction() (step 10).

Coming back to the responder, the RPC logic is all in the RpcResponderConsumer. This is not different from a specialized non-blocking consumer, as we have seen in the Consuming messages recipe, except for two details:

In this section we will discuss the use of blocking RPC and some scalability notes.

com.rabbitmq.client.RpcClient
com.rabbitmq.client.StringRpcServer

To use this recipe you will need to set up Java, Python and Ruby environments as described in the Introduction section.

To cook this recipe we are preparing four different codes:

To prepare a Java publisher:

Then to prepare a Java consumer:

The source code of the Python consumer is very similar to the Java consumer, so there is no need to repeat the needed steps. Just follow the steps of the Java consumer, looking to the source code in the archive of the recipes at:

Chapter01/Recipe06/Python_6/PyConsumer.py

In the Ruby consumer you need to use require "bunny" and then use the URI connection. Check out the source code at:

Chapter01/Recipe06/Ruby_6/RbConsumer.rb

We are now ready to mix all together, to see the recipe in action:

Both the producer and the consumers are connected to RabbitMQ with a single connection, but the logical path of the messages is depicted in the following figure:

In step 1 we have declared the exchange that we are using. The logic is the same as in the queue declaration: if the specified exchange doesn't exist, create it; otherwise, do nothing.

The second argument of exchangeDeclare() is a string, specifying the type of the exchange, fanout in this case.

In step 2 the producer sends one message to the exchange. You can just view it along with the other defined exchanges issuing the following command on the RabbitMQ command shell:

rabbitmqctl list_exchanges

The second argument in the call to channel.basicPublish() is the routing key, which is always ignored when used with a fanout exchange. The third argument, set to null, is the optional message property (more on this in the Using message properties recipe). The fourth argument is just the message itself.

When we started one consumer, it created its own temporary queue (step 9). Using the channel.queueDeclare() empty overload, we are creating a nondurable, exclusive, autodelete queue with an autogenerated name.

Launching a couple of consumers and issuing rabbitmqctl list_queues, we can see two queues, one per consumer, with their odd names, along with the persistent myFirstQueue used in previous recipes as shown in the following screenshot:

In step 5 we have bound the queues to myExchange. It is possible to monitor these bindings too, issuing the following command:

rabbitmqctl list_bindings

The monitoring is a very important aspect of AMQP; messages are routed by exchanges to the bound queues, and buffered in the queues.

The fanout exchange routes messages by just placing a copy of them in each bound queue. So, no bound queues and all the messages are just received by no one consumer (see the Handling unroutable messages recipe for more details).

As soon as we close one consumer, we implicitly destroy its private temporary queue (that's why the queues are autodelete; otherwise, these queues would be left behind unused, and the number of queues on the broker would increase indefinitely), and messages are not buffered to it anymore.

When we restart the consumer, it will create a new, independent queue and as soon as we bind it to myExchange, messages sent by the publisher will be buffered into this queue and pulled by the consumer itself.

When RabbitMQ is started for the first time, it creates some predefined exchanges. Issuing rabbitmqctl list_exchanges we can observe many existing exchanges, in addition to the one that we have defined in this recipe:

All the amq.* exchanges listed here are already defined by all the AMQP-compliant brokers and can be used instead of defining your own exchanges; they do not need to be declared at all.

We could have used amq.fanout in place of myLastnews.fanout_6, and this is a good choice for very simple applications. However, applications generally declare and use their own exchanges.

With the overload used in the recipe, the exchange is non-autodelete (won't be deleted as soon as the last client detaches it) and non-durable (won't survive server restarts). You can find more available options and overloads at http://www.rabbitmq.com/releases/rabbitmq-java-client/current-javadoc/.

To use this recipe we need to set up the Java development environment as indicated in the Introduction section.

We are going to show how to implement both the producer and the consumer, and see them in action. To implement the producer, perform the following steps:

To implement the consumer, perform the following steps:

In this recipe we have published messages (step 2) tagged with an arbitrary string (the so called routing key), to a direct exchange.

As with fanout exchanges, messages are not stored if there are no queues bound; however, in this case the consumers can choose the messages to be forwarded to these queues, depending on the binding key specified when they are bound (step 5).

Only messages with a routing key equal to the one specified in the binding will be delivered to such queues.

However, it's possible to have more queues bound with the same binding key; in this case, the broker will place a copy of the matching messages in all of them.

It is also possible to bind many different binding keys to the same queue/exchange pair, letting all the corresponding messages be delivered.

In case we deliver a message with a given routing key to an exchange, and there are no queues bound with that specific key, the message is silently dropped.

However, the producer can detect and behave consequently when this happens, as shown in detail in the Handling unroutable messages recipe.

To use this recipe we need to set up the Java development environment as indicated in the Introduction section.

Let's start with the producer:

Then, the consumers:

As in the previous recipe, messages sent to a topic exchange are tagged with a string (step 2), but it is important for a topic exchange to be composed more of dot-separated words; these are supposed to be the topics of the message. For example, in our code we have used:

technology.rabbitmq.ebook
sport.golf.paper
sport.tennis.ebook

To consume these messages the consumer has to bind myQueue to the exchange (step 5) using the appropriate key.

Using the topic exchange, the subscription/binding key specified in step 5 can be a sequence of dot-separated words and/or wildcards. AMQP wildcards are just:

So, for example:

In the last case the topic exchange behaves exactly like a fanout exchange, except for the performance, which is inevitably higher when using the former.

Again, if some messages cannot be delivered to any one queue, they are silently dropped.

The producer can detect and behave consequently when this happens, as shown in detail in the Handling unroutable messages recipe.

To use this recipe we need to set up the Java development environment as indicated in the Introduction section.

In order to guarantee that the messages have been acknowledged by the consumer after processing them, you can perform the following steps:

After we created the queue (step 1), we added the consumer to the queue and defined the ack behavior (step 2).

The parameter autoack = false informs the RabbitMQ client API that we are going to send explicit ack ourselves.

After we have got a message from the queue, we must acknowledge to RabbitMQ that we have received and properly processed the message calling channel.basicAck()(step 3). The message will be removed from the queue only when RabbitMQ receives the ack.

The method channel.basicAck() has two parameters:

The deliveryTag parameter is a value assigned by the server to the message, which you can retrieve using delivery.getEnvelope().getDeliveryTag().

If multiple is set to false the client acknowledges only the message of the deliveryTag parameter, otherwise the client acknowledges all the messages until this last one. This flag allows us to optimize consuming messages by sending ack to RabbitMQ on a block of messages instead of for each one.

Calling channel.basicAck(0,true) all the unacknowledged messages get acknowledged; the 0 stands for "all the messages".

Furthermore, calling channel.basicAck(0,false) raises an exception.

In the next chapter we will discuss the basicReject() method. This method is a RabbitMQ extension that allows further flexibility.

The Distributing messages to many consumers recipe is a real example that explains better explicit ack use.

To use this recipe we need to set up the Java development environment as indicated in the Introduction section.

In order to let two or more RabbitMQ clients properly balance consuming messages, you need to follow the given steps:

The publisher sends a message with the URL to download:

String messageUrlToDownload=  "http://www.rabbitmq.com/releases/rabbitmq-dotnet-client/v3.0.2/rabbitmq-dotnet-client-3.0.2-user-guide.pdf";
channel.basicPublish("",MyQueue,null,messageUrlToDownload.getBytes());

The consumer gets the message and downloads the referenced URL:

System.out.println("Url to download:" + messageURL);
downloadUrl(messageURL);

Once the download is terminated, the consumer sends the ack back to the broker and is ready to download the next one:

getChannel().basicAck(envelope.getDeliveryTag(),false);
System.out.println("Ack sent!");
System.out.println("Wait for the next download...");

By default, messages are heavily prefetched. Messages are retrieved by the consumers in blocks, but are actually consumed and removed from the queue when the consumers send the ack, as already seen in the previous recipe.

On the other hand, using many consumers as in this recipe, the first one will prefetch the messages, and the other consumers started later won't find any available in the queue. In order to equally distribute the work among the active consumers, we need to use channel.basicQos(1), specifying to prefetch just one message at a time.

You can find more information about load balancing in Chapter 8, Performance Tuning for RabbitMQ.

To use this recipe you will need to set up the Java development environment as indicated in the Introduction section.

In order to access the message properties you need to perform the following steps:

The AMQP message (also called content) is divided into two parts:

In step 2 we create a content header using BasicProperties:

Map<String,Object>headerMap = new HashMap<String, Object>();
BasicProperties messageProperties = new BasicProperties.Builder()
.timestamp(new Date())
.userId("guest")  
.deliveryMode(1)
.priority(1)
.headers(headerMap)
.build();

With this object we have set up the following properties:

The header is ready and we can send a message using channel.basicPublish("",myQueue, messageProperties,message.getBytes()), where messageProperties is the message header and message is the message body.

In step 4 the consumer gets a message:

public void handleDelivery(String consumerTag,Envelope envelope, 
BasicProperties properties,byte[] body) throws java.io.IOException {
System.out.println("***********message header****************");
System.out.println("Message sent at:"+ properties.getTimestamp());
System.out.println("Message sent by user:"+ properties.getUserId());
System.out.println("Message sent by App:"+properties.getAppId());
System.out.println("all properties :" + properties.toString());
System.out.println("**********message body**************");
String message = new String(body);
System.out.println("Message Body:"+message);
}

The parameter properties contains the message header and body contains its body.

Using message properties we can optimize the performance. Writing audit information or log information into the body is a typical error, because the consumer should parse the body to get them.

The body message must only contain application data (for example, a Book class), while the message properties can host other information related to the messaging mechanics or other implementation details.

For example, if the consumer wants to log when a message has been sent you can use the timestamp attribute, or if the consumer needs to distinguish a message according to a custom tag, you can put it in the headers HashMap property.

The class MessageProperties contains some pre-built BasicProperties class for standard cases. Please check the official link at http://www.rabbitmq.com/releases//rabbitmq-java-client/current-javadoc/com/rabbitmq/client/MessageProperties.html

In this example we have just used some of the properties. You can get more information at http://www.rabbitmq.com/releases//rabbitmq-java-client/current-javadoc/com/rabbitmq/client/AMQP.BasicProperties.html.

To use this recipe you will need to set up the Java development environment as indicated in the Introduction section.

You can use transactional messages by performing the following steps:

After creating a persistent queue (step 1), we have set the channel in the transaction mode using the method txSelect() (step 2). Using txCommit() the message is stored in the queue and written to the disk; the message will then be delivered to the consumer(s).

The method txSelect()must be called at least once before txCommit() or txRollback().

As in a DBMS you can use a rollback method. In the following case the message isn't stored or delivered:

channel.basicPublish("",myQueue, MessageProperties.PERSISTENT_TEXT_PLAIN ,message.getBytes());
channel.txRollback(); 

The transactions can reduce the application's performance, because the broker doesn't cache the messages and the tx operations are synchronous.

In the next chapter we will discuss the publish confirm plugin, which is a faster way to get the confirmation for the operations.

To use this recipe you will need to set up the Java development environment as indicated in the Introduction section.

In order to handle unroutable messages, you need to perform the following steps:

When we execute the publisher, the messages sent to myExchange won't reach any destination since it has no bound queues. However, these messages aren't, they are redirected to an internal queue. The HandlingReturnListener class will handle such messages using handleReturn().

The ReturnListener class is bound to a publisher channel, and it will trap only its own unroutable messages.

You can also find a consumer in the source code example. Try also to execute the publisher and the consumer together, and then stop the consumer.

If you don't set the channel ReturnListener, the unroutable messages are silently dropped by the broker. In case you want to be notified about the unroutable messages, it's important to set the mandatory flag to true; if false, the unroutable messages are dropped as well.