RabbitMQ JMS Client

The topic selector plugin is included with RabbitMQ. Like any RabbitMQ plugin, you need to enable the plugin in order to use it.

Enable the plugin using the rabbitmq-plugins command:

You don’t need to restart the broker to activate the plugin.

Use your favorite build management tool to add the client dependencies to your project.

To enable the JMS Client in a Java container (e.g. Java EE application server, web container), you must install the JMS client JAR files and its dependencies in the container and then define JMS resources in the container’s naming system so that JMS clients can look them up. The methods for accomplishing these tasks are container-specific, please refer to the vendors’ documentation.

For standalone applications, you need to add the JMS client JAR files and its dependencies to the application classpath. The JMS resources can be defined programmatically or through a dependency injection framework like Spring.

To define the JMS ConnectionFactory in JNDI, e.g. in Tomcat:

To define the JMS ConnectionFactory in JNDI, e.g. in WildFly (as of JMS Client 1.7.0):

Here is the equivalent Spring bean example (Java configuration):

And here is the Spring XML configuration:

The following table lists all of the attributes/properties that are available.

A JMS destination can be defined so that a JMS application can send Messages to a predefined RabbitMQ 'destination' (exchange/routing key) using the JMS API in the normal way. The messages are written "in the clear," which means that any AMQP 0-9-1 client can read them without having to understand the internal format of Java JMS messages. Only BytesMessages and TextMessages can be written in this way.

When messages are sent to an 'amqp' Destination, JMS message properties are mapped onto AMQP 0-9-1 headers and properties as appropriate. For example, the JMSPriority property converts to the priority property for the AMQP 0-9-1 message. (It is also set as a header with the name "JMSPriority".) User-defined properties are set as named message header values, provided they are boolean, numeric or String types.

Similarly, a JMS destination can be defined that reads messages from a predefined RabbitMQ queue. A JMS application can then read these messages using the JMS API. RabbitMQ JMS Client packs them up into JMS Messages automatically. Messages read in this way are, by default, BytesMessages, but individual messages can be marked TextMessage (by adding an AMQP message header called "JMSType" whose value is "TextMessage"), which will interpret the byte-array payload as a UTF8 encoded String and return them as TextMessages.

When reading from an 'amqp' Destination, values are mapped back to JMS message properties, except that any explicit JMS property set as a message header overrides the natural AMQP 0-9-1 header value, unless this would misrepresent the message. For example, JMSDeliveryMode cannot be overridden in this way.

The com.rabbitmq.jms.admin package contains the RMQDestination class, which implements Destination in the JMS interface. This is extended with a new constructor:

This constructor creates a destination for JMS for RabbitMQ mapped onto an AMQP 0-9-1 resource. The parameters are the following:

Applications that declare destinations in this way can use them directly, or store them in a JNDI provider for JMS applications to retrieve. Such destinations are non-temporary, queue destinations.

The RMQDestination object has the following new instance fields:

There are getters and setters for these fields, which means that a JNDI <Resource/> definition or an XML Spring bean definition can use them, for example JNDI with Tomcat:

This is the equivalent with WildFly (as of JMS Client 1.7.0):

This is the equivalent Spring bean example (Java configuration):

And here is the Spring XML configuration:

Following is a complete list of the attributes/properties that are available:

An RPC client can be implemented in pure JMS like the following:

It’s also possible to create a single reply-to destination instead of a temporary destination for each request. This is more efficient but requires to properly correlate the response with the request, by using e.g. a correlation ID header. RabbitMQ’s direct reply-to is another alternative (see below).

Note this sample uses a MessageListener and a BlockingQueue to wait for the response. This implies a network roundtrip to register an AMQP consumer and another one to close the consumer. MessageConsumer#receive could have been used as well, in this case the JMS client internally polls the reply destination to get the response, which can result in several network roundtrips if the response takes some time to come. The request call will also incur a constant penalty equals to the polling interval (100 milliseconds by default).

The server part looks like the following:

Note the connectionFactory.setDeclareReplyToDestination(false) statement: it is necessary when using temporary reply-to destinations. If this flag is not set to false on the RPC server side, the JMS client will try to re-create the temporary reply-to destination, which will interfere with the client-side declaration.

See this test for a full RPC example.

The JMS client also supports direct reply-to, which is faster as it doesn’t imply creating a temporary reply destination:

Using direct reply-to for JMS-based RPC has the following implications:

See this test for a full RPC example using direct reply-to.

Spring JMS is a popular way to work with JMS as it avoids most of JMS boilerplate.

The following sample shows how a client can perform RPC with the JmsTemplate:

This is no different from any other JMS client.

The JmsTemplate uses a temporary reply-to destination, so the call to connectionFactory.setDeclareReplyToDestination(false) on the RPC server side is necessary, just like with regular JMS.

RPC with direct reply-to must be implemented with a SessionCallback, as the reply destination must be explicitly declared:

See this test for a full example of RPC with Spring JMS, including using a @JmsListener bean for the server part.

JMS topics are implemented using an AMQP topic exchange and a dedicated AMQP queue for each JMS topic subscriber. The AMQP topic exchange is jms.temp.topic or jms.durable.topic, depending on whether the JMS topic is temporary or not, respectively. Let’s take an example with a subscription to a durable my.jms.topic JMS topic:

If another subscriber subscribes to my.jms.topic, it will have its own AMQP queue and both subscribers will receive messages published to the jms.durable.topic exchange with the my.jms.topic routing key.

The example above assumes no topic selector is used when declaring the subscribers. If a topic selector is in use, a x-jms-topic-typed exchange will sit between the jms.durable.topic topic exchange and the subscriber queue. So the topology is the following when subscribing to a durable my.jms.topic JMS topic with a selector:

Exchanges can be bound together thanks to a RabbitMQ extension. Note the Topic Selector Plugin must be enabled for topic selectors to work.

Prior to version 1.2.0 of the JMS client, in client acknowledgement mode (Session.CLIENT_ACKNOWLEDGE), acknowledging any message from an open session would acknowledge every unacknowledged message of that session, whether they were received before or after the message being acknowledged.

Currently, the behaviour of Session.CLIENT_ACKNOWLEDGE mode is modified so that, when calling msg.acknowledge(), only the message msg and all previously received unacknowledged messages on that session are acknowledged. Messages received after msg was received are not affected. This is a form of group acknowledgement, which differs slightly from the JMS specification but is likely to be more useful, and is compatible with the vast majority of uses of the existing acknowledge function.

For even finer control, a new acknowledgement mode may be set when creating a session, called RMQSession.CLIENT_INDIVIDUAL_ACKNOWLEDGE.

A session created with this acknowledgement mode will mean that messages received on that session will be acknowledged individually. That is, the call msg.acknowledge() will acknowledge only the message msg and not affect any other messages of that session.

The acknowledgement mode RMQSession.CLIENT_INDIVIDUAL_ACKNOWLEDGE is equivalent to Session.CLIENT_ACKNOWLEDGE in all other respects. In particular the getAcknowledgeMode() method returns Session.CLIENT_ACKNOWLEDGE even if RMQSession.CLIENT_INDIVIDUAL_ACKNOWLEDGE has been set.

Any instance of a class that implements the javax.jms.Message interface can be sent by a JMS message producer.

All properties of the message required by send() are correctly interpreted except that the JMSReplyTo header and objects (as property values or the body of an ObjectMessage) that cannot be deserialized are ignored.

The implementation extracts the properties and body from the Message instance using interface methods and recreates it as a message of the right (RMQMessage) type (BytesMessage, MapMessage, ObjectMessage, TextMessage, or StreamMessage) before sending it. This means that there is some performance loss due to the copying; but in the normal case, when the message is an instance of com.rabbitmq.jms.client.RMQMessage, no copying is done.