Java Client API Guide
This page gives an overview of the RabbitMQ Java client API.
The code samples given here demonstrate connecting to AMQP brokers and using RPC services exposed via AMQP.
For more details, please see the relevant Javadoc documentation.
The client API is closely modelled on the AMQP protocol specification, with additional abstractions for ease of use.
Protocol class overview
The holder class AMQP stores all the code generated automatically from the AMQP XML protocol definition specification. It contains all required content-class-specific content header definitions (such asAMQP.BasicProperties) and all the method request and response descriptors (such asAMQP.Basic.Publish and AMQP.Queue.BindOk), as well as useful protocol-specific constants and other values.
import com.rabbitmq.client.AMQP;
The method request and response descriptors (and the BasicProperties class) come with Builder classes (following the Builder Pattern) to make constructing protocol objects easier and to allow us to construct them with immutable state.
We illustrate the builder classes by constructing some AMQP.BasicProperties objects with its Builderclass:
AMQP.BasicProperties.Builder bob = new AMQP.BasicProperties.Builder();
AMQP.BasicProperties minBasic = bob.build();
AMQP.BasicProperties minPersistentBasic = bob.deliveryMode(2).build();
AMQP.BasicProperties persistentBasic
= bob.priority(0).contentType("application/octet-stream").build();
AMQP.BasicProperties persistentTextPlain = bob.contentType("text/plain").build();
bob (the builder) is constructed first and whenever build() is invoked this method returns a newBasicProperties object with the properties set in bob at that point. The parameter calls (deliveryMode,priority, etc.) update bob and not the BasicProperties object. Each of these methods returns the updated builder (allowing chaining, for example as in persistentBasic above). This pattern allows the parameters to be named and set in any order, the consistency of the complete set of parameters to be checked at build() time, the built objects to be immutable, and partially initialised builders to be re-used.
For details and exact definitions of the AMQP protocol, please see the AMQP specification document; for details of the API (including the Builder classes) see the Javadoc documentation; and for background on the Builder Pattern see this Dr. Dobb’s article or refer to Effective Java by Joshua Bloch.
Connections and channels
The core API classes are Connection and Channel, representing an AMQP connection and an AMQP data channel, respectively:
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
Connecting to a broker
The following code connects to an AMQP broker using the given parameters (host name, port number, etc):
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername(userName);
factory.setPassword(password);
factory.setVirtualHost(virtualHost);
factory.setHost(hostName);
factory.setPort(portNumber);
Connection conn = factory.newConnection();
All of these parameters have sensible defaults for a RabbitMQ server running locally.
Alternatively, AMQP URIs may be used:
ConnectionFactory factory = new ConnectionFactory();
factory.setUri("amqp://userName:password@hostName:portNumber/virtualHost");
Connection conn = factory.newConnection();
All of these parameters have sensible defaults for a stock RabbitMQ server running locally.
The Connection interface can then be used to open a channel:
Channel channel = conn.createChannel();
The channel can now be used to send and receive messages, as described in subsequent sections.
To disconnect, simply close the channel and the connection:
channel.close();
conn.close();
Note that closing the channel may be considered good practice, but isn’t strictly necessary here - it will be done automatically anyway when the underlying connection is closed.
Advanced Connection options
Consumer thread pool
Consumer threads (see Receiving below) are automatically allocated in a new ExecutorService thread pool by default. If greater control is required supply an ExecutorService on the newConnection() method, so that this pool of threads is used instead. Here is an example where a larger thread pool is supplied than is normally allocated:
ExecutorService es = Executors.newFixedThreadPool(20);
Connection conn = factory.newConnection(es);
Both
Executors and
ExecutorService classes are in the
java.util.concurrent package.
When the connection is closed a default ExecutorService will be shutdown(), but a user-suppliedExecutorService (like es above) will not be shutdown(). Clients that supply a custom ExecutorServicemust ensure it is shutdown eventually (by calling its shutdown() method), or else the pool’s threads may prevent JVM termination.
The same executor service may be shared between multiple connections, or serially re-used on re-connection but it cannot be used after it is shutdown().
Use of this feature should only be considered if there is evidence that there is a severe bottleneck in the processing of Consumer callbacks. If there are no Consumer callbacks executed, or very few, the default allocation is more than sufficient. The overhead is initially minimal and the total thread resources allocated are bounded, even if a burst of consumer activity may occasionally occur.
Address array
It is possible to pass an Address array to newConnection(). An Address is simply a convenience class (in the com.rabbitmq.client package) with host and port components. For example:
Address[] addrArr = new Address[]{ new Address(hostname1, portnumber1)
, new Address(hostname2, portnumber2)};
Connection conn = factory.newConnection(addrArr);
will attempt to connect to
hostname1:portnumber1, and if that fails to
hostname2:portnumber2. The connection returned is the first in the array that succeeds (without throwing
IOException). This is entirely equivalent to repeatedly setting host and port on a factory, calling
factory.newConnection() each time, until one of them succeeds.
If an ExecutorService is provided as well (using the form factory.newConnection(es, addrArr)) the thread pool is associated with the (first) successful connection.
Using exchanges and queues
Client applications work with exchanges and queues, the high-level building blocks of AMQP. These must be "declared" before they can be used. Declaring either type of object simply ensures that one of that name exists, creating it if necessary.
Continuing the previous example, the following code declares an exchange and a queue, then binds them together.
channel.exchangeDeclare(exchangeName, "direct", true);
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, exchangeName, routingKey);
This will actively declare the following objects, both of which can be customised by using additional parameters. Here neither of them have any special arguments.
- a durable, non-autodelete exchange of "direct" type
- a non-durable, exclusive, autodelete queue with a generated name
The above function calls then bind the queue to the exchange with the given routing key.
Note that this would be a typical way to declare a queue when only one client wants to work with it: it doesn’t need a well-known name, no other client can use it (exclusive) and will be cleaned up automatically (autodelete). If several clients want to share a queue with a well-known name, this code would be appropriate:
channel.exchangeDeclare(exchangeName, "direct", true);
channel.queueDeclare(queueName, true, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
This will actively declare:
- a durable, non-autodelete exchange of "direct" type
- a durable, non-exclusive, non-autodelete queue with a well-known name
Note that all of these Channel API methods are overloaded. These convenient short forms ofexchangeDeclare, queueDeclare and queueBind use sensible defaults. There are also longer forms with more parameters, to let you override these defaults as necessary, giving full control where needed.
This "short form, long form" pattern is used throughout the client API uses.
Publishing messages
To publish a message to an exchange, use Channel.basicPublish as follows:
byte[] messageBodyBytes = "Hello, world!".getBytes();
channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);
For fine control, you can use overloaded variants to specify the mandatory flag, or send messages with pre-set message properties:
channel.basicPublish(exchangeName, routingKey, mandatory,
MessageProperties.PERSISTENT_TEXT_PLAIN,
messageBodyBytes);This sends a message with delivery mode 2 (persistent), priority 0 and content-type "text/plain". You can build your own message properties object, using a Builder class mentioning as many properties as you like, for example:
channel.basicPublish(exchangeName, routingKey,
new AMQP.BasicProperties.Builder()
.contentType("text/plain").deliveryMode(2)
.priority(1).userId("bob")
.build()),
messageBodyBytes);We have not illustrated all the possibilities here.
Note that BasicProperties is an inner class of the autogenerated holder class AMQP.
Channel thread-safety
Channel instances are safe for use by multiple threads. Requests into a Channel are serialized, with only one thread being able to run a command on the Channel at a time. Even so, applications should prefer using aChannel per thread instead of sharing the same Channel across multiple threads.
Receiving messages by subscription
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
The most efficient way to receive messages is to set up a subscription using the Consumer interface. The messages will then be delivered automatically as they arrive, rather than having to be explicitly requested.
When calling the API methods relating to Consumers, individual subscriptions are always referred to by their consumer tags, which can be either client- or server-generated as explained in the AMQP specification document. Distinct Consumers on the same Channel must have distinct consumer tags.
The easiest way to implement a Consumer is to subclass the convenience class DefaultConsumer. An object of this subclass can be passed on a basicConsume call to set up the subscription:
boolean autoAck = false;
channel.basicConsume(queueName, autoAck, "myConsumerTag",
new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException
{
String routingKey = envelope.getRoutingKey();
String contentType = properties.contentType;
long deliveryTag = envelope.getDeliveryTag();
// (process the message components here ...)
channel.basicAck(deliveryTag, false);
}
});Here, since we specified autoAck = false, it is necessary to acknowledge messages delivered to theConsumer, most conveniently done in the handleDelivery method, as illustrated.
More sophisticated Consumers will need to override further methods. In particular, handleShutdownSignal is called when channels and connections close, and handleConsumeOk is passed the consumer tag before any other callbacks to that Consumer are called.
Consumers can also implement the handleCancelOk and handleCancel methods to be notified of explicit and implicit cancellations, respectively.
You can explicitly cancel a particular Consumer with Channel.basicCancel:
channel.basicCancel(consumerTag);
passing the consumer tag.
Callbacks to Consumers are dispatched on a thread separate from the thread managed by the Connection. This means that Consumers can safely call blocking methods on the Connection or Channel, such asqueueDeclare, txCommit, basicCancel or basicPublish.
Each Channel has its own dispatch thread. For the most common use case of one Consumer per Channel, this means Consumers do not hold up other Consumers. If you have multiple Consumers per Channel be aware that a long-running Consumer may hold up dispatch of callbacks to other Consumers on that Channel.
Retrieving individual messages
To explicitly retrieve messages, use Channel.basicGet. The returned value is an instance of GetResponse, from which the header information (properties) and message body can be extracted:
boolean autoAck = false;
GetResponse response = channel.basicGet(queueName, autoAck);
if (response == null) {
// No message retrieved.
} else {
AMQP.BasicProperties props = response.getProps();
byte[] body = response.getBody();
long deliveryTag = response.getEnvelope().getDeliveryTag();
...and since the autoAck = false above, you must also call Channel.basicAck to acknowledge that you have successfully received the message:
...
channel.basicAck(method.deliveryTag, false); // acknowledge receipt of the message
}Handling unroutable messages
If a message is published with the "mandatory" flags set, but cannot be routed, the broker will return it to the sending client (via a AMQP.Basic.Return command).
To be notified of such returns, clients can implement the ReturnListener interface and callChannel.setReturnListener. If the client has not configured a return listener for a particular channel, then the associated returned messages will be silently dropped.
channel.setReturnListener(new ReturnListener() {
public void handleBasicReturn(int replyCode,
String replyText,
String exchange,
String routingKey,
AMQP.BasicProperties properties,
byte[] body)
throws IOException {
...
}
});A return listener will be called, for example, if the client publishes a message with the "mandatory" flag set to an exchange of "direct" type which is not bound to a queue.
Basic RPC
As a programming convenience, the Java client API offers a class RpcClient which uses a temporary reply queue to provide simple RPC-style communication facilities via AMQP.
The class doesn’t impose any particular format on the RPC arguments and return values. It simply provides a mechanism for sending a message to a given exchange with a particular routing key, and waiting for a response on a reply queue.
import com.rabbitmq.client.RpcClient;
RpcClient rpc = new RpcClient(channel, exchangeName, routingKey);
(The implementation details of how this class uses AMQP are as follows: request messages are sent with thebasic.correlation_id field set to a value unique for this RpcClient instance, and with basic.reply_toset to the name of the reply queue.)
Once you have created an instance of this class, you can use it to send RPC requests by using any of the following methods:
byte[] primitiveCall(byte[] message);
String stringCall(String message)
Map mapCall(Map message)
Map mapCall(Object[] keyValuePairs)
The primitiveCall method transfers raw byte arrays as the request and response bodies. The methodstringCall is a thin convenience wrapper around primitiveCall, treating the message bodies as Stringinstances in the default character encoding.
The mapCall variants are a little more sophisticated: they encode a java.util.Map containing ordinary Java values into an AMQP binary table representation, and decode the response in the same way. (Note that there are some restrictions on what value types can be used here - see the javadoc for details.)
All the marshalling/unmarshalling convenience methods use primitiveCall as a transport mechanism, and just provide a wrapping layer on top of it.
Shutdown Protocol
Overview of the AMQP client shutdown
The AMQP connection and channel share the same general approach to managing network failure, internal failure, and explicit local shutdown.
The AMQP connection and channel have the following lifecycle states:
- open: the object is ready to use
- closing: the object has been explicitly notified to shut down locally, has issued a shutdown request to any supporting lower-layer objects, and is waiting for their shutdown procedures to complete
- closed: the object has received all shutdown-complete notification(s) from any lower-layer objects, and as a consequence has shut itself down
Those objects always end up in the closed state, regardless of the reason that casued the closure, like an application request, an internal client library failure, a remote network request or network failure.
The AMQP connection and channel objects possess the following shutdown-related methods:
- addShutdownListener(ShutdownListener listener) and removeShutdownListener(ShutdownListener listener), to manage any listeners, which will be fired when the object transitions to closed state. Note that, adding a ShutdownListener to an object that is already closed will fire the listener immediately
- getCloseReason(), to allow the investigation of what was the reason of the object’s shutdown
- isOpen(), useful for testing whether the object is in an open state
- close(int closeCode, String closeMessage), to explictly notify the object to shut down
Simple usage of listeners would look like:
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.ShutdownListener;
connection.addShutdownListener(new ShutdownListener() {
public void shutdownCompleted(ShutdownSignalException cause)
{
...
}
});Information about the circumstances of a shutdown
One can retrieve the ShutdownSignalException, which contains all the information available about the close reason, either by explictly calling the getCloseReason() method or by using the cause parameter in the service(ShutdownSignalException cause) method of the ShutdownListener class.
The ShutdownSignalException class provides methods to analyze the reason of the shutdown. By calling the isHardError() method we get information whether it was a connection or a channel error.
public void shutdownCompleted(ShutdownSignalException cause)
{
if (cause.isHardError())
{
Connection conn = (Connection)cause.getReference();
if (!cause.isInitiatedByApplication())
{
Object reason = cause.getReason();
...
}
...
} else {
Channel ch = (Channel)cause.getReference();
...
}
}Atomicity and use of the isOpen() method
Use of the isOpen() method of channel and connection objects is not recommended for production code, because the value returned by the method is dependent on the existence of the shutdown cause. The following code illustrates the possibility of race conditions:
public void brokenMethod(Channel channel)
{
if (channel.isOpen())
{
// The following code depends on the channel being in open state.
// However there is a possibility of the change in the channel state
// between isOpen() and basicQos(1) call
...
channel.basicQos(1);
}
}Instead, we should normally ignore such checking, and simply attempt the action desired. If during the execution of the code the channel of the connection is closed, a ShutdownSignalException will be thrown indicating that the object is in an invalid state. We should also catch for IOException caused either bySocketException, when broker closes the connection unexpectedly, or ShutdownSignalException, when broker initiated clean close.
public void validMethod(Channel channel)
{
try {
...
channel.basicQos(1);
} catch (ShutdownSignalException sse) {
// possibly check if channel was closed
// by the time we started action and reasons for
// closing it
...
} catch (IOException ioe) {
// check why connection was closed
...
}
}Bundled examples
The Java client library distribution is shipped with a rabbitmq-client-tests.jar, containing several small example programs for exercising the functionality of the RabbitMQ server. The source code for these examples is in the test/src folder in the source distribution.
The script runjava.{sh,bat} runs Java with the class path correctly configured for the examples, e.g.runjava.sh com.rabbitmq.examples.TestMain runs the TestMain functional tests.
AMQP Protocol Capture/Analysis Tool
There's a very basic, very simple AMQP protocol analyzer in class com.rabbitmq.tools.Tracer. Invoke it with
runjava.sh com.rabbitmq.tools.Tracer listenPort connectHost connectPort
- listenPort
- port to listen for incoming AMQP connections on - defaults to 5673.
- connectHost
- hostname to use when making an outbound connection in response to an incoming connection - defaults to localhost.
- connectPort
- port number to use when making an outbound connection - defaults to 5672.
![FanoutExchange_thumb[2]](http://aspiringcraftsman.com/wp-content/uploads/2012/03/FanoutExchange_thumb2.png "FanoutExchange_thumb[2]")
![TopicExchange_thumb[2]](http://aspiringcraftsman.com/wp-content/uploads/2012/03/TopicExchange_thumb2.png "TopicExchange_thumb[2]")
![HeadersExchange_thumb[2]](http://aspiringcraftsman.com/wp-content/uploads/2012/03/HeadersExchange_thumb2.png "HeadersExchange_thumb[2]")
