In this recipe we are learning how to send a message to an AMQP queue. We will be introduced to the building blocks of AMQP messaging: messages, queues, and exchanges.
You can find the source at Chapter01/Recipe02/src/rmqexample.
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:
Declare the queue, calling the
queueDeclare()method oncom.rabbitmq.client.Channel:String myQueue = "myFirstQueue"; channel.queueDeclare(myQueue, true, false, false, null);
Send the very first message to the RabbitMQ broker:
String message = "My message to myFirstQueue"; channel.basicPublish("",myQueue, null, message.getBytes());Send the second message with different options:
channel.basicPublish("",myQueue,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
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
If the queue already exists but has been created with different parameters, queueDeclare() will raise an exception.
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:
queue: This is just the name of the queue where we will be storing the messages.durable: This specifies whether the queue will survive server restarts. Note that it is required for a queue to be declared as durable if you want persistent messages to survive a server restart.exclusive: This specifies whether the queue is restricted to only this connection.autoDelete: This specifies whether the queue will be automatically deleted by the RabbitMQ broker as soon as it is not in use.arguments: This is an optional map of queue construction arguments.
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.
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 restartOn Linux, Ubuntu, Debian, and so on:
/etc/init.d/rabbitmq restartOn Windows:
sc stop rabbitmq / sc start rabbitmq
How many messages should we expect when we run rabbitmqclt list_queues again?
In order to be sure that a specific queue already exists, replace channel.queueDeclare() with channel.queueDeclarePassive(). The behavior of the two methods is the same in case the queue already exists; but in case it doesn't, the first one will silently create it and return back (that's actually the most frequently used case), the latter will raise an exception.