JMS queues with JEE6 and JBoss 7

This post will show how to configure a JMS queue on JBoss 7.1.1-Final and how to make use of it through a JEE 6 application. Besides, some aspects regarding transactional session and JMS queue’s number of consumers will also be presented.

JMS

This post does not aim to depict JMS in details, but it will give an overview about it.
JMS is part of the Java EE specification and can be considered a MOM (Message Oriented Middleware) which allows clients to exchange asynchronous messages among them.
There are basically two models to exchange messages:

• Queue: It’s a point-to-point on which one side (producer) pushes a message and places it in a JMS queue and the other side (consumer) pulls the message from the queue. This post will be using this model throughout its example.
• Topic: Adopts the publish-subscribe model on which one side (publisher) pushes as message on a JSM topic and, for this topic, there can be 0 or more subscribers that receive all the published messages.

More details can be found on the JEE 6 Tutorial.

Configuring a JMS queue on JBoss

JBoss AS uses the HornetQ library (http://www.jboss.org/hornetq) as its JMS provider. HornetQ is an open-source messaging system, fully compliant with JMS specification, with characteristics of robustness, performance and scalability. In order to configure a JMS queue, it’s necessary to edit the JBOSS_HOME/standalone/configuration/standalone.xml file which, as default, does not come with the JMS module configured. (If you prefer to use the configuration file standalone-full.xml, that’s OK, as it already comes with the JMS configured.)
Open the JBOSS_HOME/standalone/configuration/standalone.xml file and edit the following sections:

Add the messaging module inside <extensions>

<extensions>
	...
	<extension module="org.jboss.as.messaging"/>
</extensions>

Add the messaging subsystem configuration right bellow <subsystem xmlns=”urn:jboss:domain:weld:1.0″/>

        <subsystem xmlns="urn:jboss:domain:messaging:1.1">
            <hornetq-server>
                <persistence-enabled>true</persistence-enabled>
                <journal-file-size>102400</journal-file-size>
                <journal-min-files>2</journal-min-files>

                <connectors>
                    <netty-connector name="netty" socket-binding="messaging"/>
                    <netty-connector name="netty-throughput" socket-binding="messaging-throughput">
                        <param key="batch-delay" value="50"/>
                    </netty-connector>
                    <in-vm-connector name="in-vm" server-id="0"/>
                </connectors>

                <acceptors>
                    <netty-acceptor name="netty" socket-binding="messaging"/>
                    <netty-acceptor name="netty-throughput" socket-binding="messaging-throughput">
                        <param key="batch-delay" value="50"/>
                        <param key="direct-deliver" value="false"/>
                    </netty-acceptor>
                    <in-vm-acceptor name="in-vm" server-id="0"/>
                </acceptors>

                <security-settings>
                    <security-setting match="#">
                        <permission type="send" roles="guest"/>
                        <permission type="consume" roles="guest"/>
                        <permission type="createNonDurableQueue" roles="guest"/>
                        <permission type="deleteNonDurableQueue" roles="guest"/>
                    </security-setting>
                </security-settings>

                <address-settings>
                    <address-setting match="#">
                        <dead-letter-address>jms.queue.DLQ</dead-letter-address>
                        <expiry-address>jms.queue.ExpiryQueue</expiry-address>
                        <redelivery-delay>0</redelivery-delay>
                        <max-size-bytes>10485760</max-size-bytes>
                        <address-full-policy>BLOCK</address-full-policy>
                        <message-counter-history-day-limit>10</message-counter-history-day-limit>
                    </address-setting>
                </address-settings>

                <jms-connection-factories>
                    <connection-factory name="InVmConnectionFactory">
                        <connectors>
                            <connector-ref connector-name="in-vm"/>
                        </connectors>
                        <entries>
                            <entry name="java:/ConnectionFactory"/>
                        </entries>
                    </connection-factory>
                    <connection-factory name="RemoteConnectionFactory">
                        <connectors>
                            <connector-ref connector-name="netty"/>
                        </connectors>
                        <entries>
                            <entry name="RemoteConnectionFactory"/>
                            <entry name="java:jboss/exported/jms/RemoteConnectionFactory"/>
                        </entries>
                    </connection-factory>
                    <pooled-connection-factory name="hornetq-ra">
                        <transaction mode="xa"/>
                        <connectors>
                            <connector-ref connector-name="in-vm"/>
                        </connectors>
                        <entries>
                            <entry name="java:/JmsXA"/>
                        </entries>
                    </pooled-connection-factory>
                </jms-connection-factories>

                <jms-destinations>
                    <jms-queue name="testQueue">
                        <entry name="queue/test"/>
                        <entry name="java:jboss/exported/jms/queue/test"/>
                    </jms-queue>
                    <jms-topic name="testTopic">
                        <entry name="topic/test"/>
                        <entry name="java:jboss/exported/jms/topic/test"/>
                    </jms-topic>
                </jms-destinations>
            </hornetq-server>
        </subsystem>

This is the default configuration and the code example will be using the testQueue queue, which is already pre-configured. If you wish to add a new queue, you just need to add a new <jms-queue>.
In order to access a queue, we need a ConnectionFactory. Our configuration has 2 of them (one regular and on transactional), which can be be accessed by java:/ConnectionFactory and java:/JmsXA names respectively and a queue can be accessed by either java:/queue/test or java:jboss/exported/jms/queue/test names.

More details regarding configuration can be found in JBoss and HornetQ documentation.

It’s also needed to configure the socket-binding referenced by <connectors> and <acceptors>. Scroll down until the section <socket-binding-group> and add the following entries:

...
        <socket-binding name="messaging" port="5445"/>
        <socket-binding name="messaging-throughput" port="5455"/>
...

To finalize JBoss configuration, it’s needed to define the instances pool to be used by the message consumer. Go until the <subsystem xmlns=”urn:jboss:domain:ejb3:1.2″> section and add the following content between </session-bean> and <pools>:

            <mdb>
                <resource-adapter-ref resource-adapter-name="hornetq-ra"/>
                <bean-instance-pool-ref pool-name="mdb-strict-max-pool"/>
            </mdb>

Now, the JBoss is fully configured and we can start to code the example.

Demo application

Our application will be fairly simple and will be packaged as a .war file. It will comprise:

• A simple POJO (User) with a String attribute called “name”
• A Servlet responsible for creating a message (User) and sending a it to an EJB that will send it to the JMS queue
• A MDB responsible for receiving messages from the testQueue queue

We will be using maven as build tool, so you can choose your preferable IDE.

The pom.xml file should be configured as showed bellow:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<groupId>com.lucianomolinari</groupId>
	<artifactId>jms_example</artifactId>
	<version>1.0.0</version>
	<packaging>war</packaging>

	<build>
		<plugins>
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-compiler-plugin</artifactId>
				<version>2.3.2</version>
				<configuration>
					<source>1.6</source>
					<target>1.6</target>
				</configuration>
			</plugin>
			<plugin>
				<artifactId>maven-war-plugin</artifactId>
				<version>2.4</version>
				<configuration>
					<failOnMissingWebXml>false</failOnMissingWebXml>
				</configuration>
			</plugin>
		</plugins>
	</build>

	<dependencies>
		<dependency>
			<groupId>javax</groupId>
			<artifactId>javaee-api</artifactId>
			<version>6.0</version>
			<scope>provided</scope>
		</dependency>
	</dependencies>

</project>

POJO that will be sent as the JMS message:

package com.lucianomolinari.jms_example.entity;

import java.io.Serializable;

public final class User implements Serializable {
	private static final long serialVersionUID = 8734596722276424601L;

	private final String name;

	public User(String name) {
		this.name = name;
	}

	public String getName() {
		return name;
	}

}

EJB responsible for sending the JMS message:

package com.lucianomolinari.jms_example.jms;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import javax.ejb.Stateless;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import com.lucianomolinari.jms_example.entity.User;

@Stateless
public class MessageCreator {
	@Resource(mappedName = "java:/ConnectionFactory")
	private ConnectionFactory connectionFactory;

	@Resource(mappedName = "java:/queue/test")
	private Destination destination;

	private Connection connection;
	private Session session;
	private MessageProducer messageProducer;

	@PostConstruct
	public void init() {
		try {
			connection = connectionFactory.createConnection();
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			messageProducer = session.createProducer(destination);
		} catch (JMSException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		}
	}

	@PreDestroy
	public void destroy() {
		if (connection != null) {
			try {
				connection.close();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
	}

	public void sendMessage(User user) {
		ObjectMessage message;
		try {
			message = session.createObjectMessage(user);
			messageProducer.send(message);
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

}

MDB that will receive the message from the queue:

package com.lucianomolinari.jms_example.jms;

import javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;

import com.lucianomolinari.jms_example.entity.User;

@MessageDriven(activationConfig = {
		@ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
		@ActivationConfigProperty(propertyName = "destination", propertyValue = "java:/queue/test") })
public class MessageConsumer implements MessageListener {

	@Override
	public void onMessage(Message message) {
		ObjectMessage objMsg = (ObjectMessage) message;
		try {
			System.out.println(">>>>> Receiving message with user " + ((User) objMsg.getObject()).getName());
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

}

Finally the Servlet responsible for invoking the EJB producer:

package com.lucianomolinari.jms_example.web.servlet;

import java.io.IOException;

import javax.ejb.EJB;
import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import com.lucianomolinari.jms_example.entity.User;
import com.lucianomolinari.jms_example.jms.MessageCreator;

@WebServlet("/test")
public class ServletCreateMessage extends HttpServlet {
	private static final long serialVersionUID = -973490316445738120L;

	@EJB
	private MessageCreator messageCreator;

	@Override
	protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
		System.out.println(">>>>Creating test message...");
		messageCreator.sendMessage(new User("Test user"));
	}

}

In order to send a JMS message, just access the URL http://localhost:8080/jms_example-1.0.0/test. It is possible to check the message being produced and consumed through JBoss log.

Important considerations

Transactional session

When a JMS session is created, the first parameter (boolean) indicates whether that session should be transactional or not. In the example above, a non-transactional session was created:

session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

However, that’s not always the desired behaviour, as in many cases the JMS message sending is part of a bigger transaction, that involves other tasks, such as updating a database. In order to make it a transactional session, it’s just needed to inform true as the parameter value:

session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);

Unfortunately, if you make only this change and run the application again, you’ll realize that the message never gets sent/consumed to/from the queue. That happens because the ConnectionFactory used (java:/ConnectionFactory) has not been configured to be transactional. In order to make the message to be sent/consumed property with a transactional session, there are 2 approaches:

Commit explicitly

One approach is to manually commit the session after the message sending:

	public void sendMessage(User user) {
		ObjectMessage message;
		try {
			message = session.createObjectMessage(user);
			messageProducer.send(message);
			session.commit();
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

However, this approach is not the recommended one, by the same reason it’s not recommended to use a non-transactional session: the message sending may be part of a broader transaction.

Make use of a ConnectionFactory with transactional support

There’s another ConnectionFactory created on JBoss (java:/JmsXA) that is configured as transactional (<transaction mode=”xa”/>). So, we can get the initial code and perform the following changes:

Update ConnectionFactory’s mappedName to para java:/JmsXA

	@Resource(mappedName = "java:/JmsXA")
	private ConnectionFactory connectionFactory;

And make the session transactional

session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);

With that, everything starts to work properly again and better: now our sending message process can be part of a bigger transaction.

Configuring the number of consumers for a queue

To finish this post, a simple but powerful configuration will be described. This configuration can be really useful for helping your application to achieve better performance/scalability regarding consuming messages from a JMS queue.

Imagine the following scenario:

• The processing required to process a message is fairly heavy and it takes 5 seconds to be completed
• The number of messages per second placed on the queue is “reasonable” (10 messages/second)
• There are only 5 MDB instances consuming the messages

Taking into account a 10 seconds window time, your queue would have received 100 messages and your application would have consumed only 10 of these messages, once each MDB instance would have consumed 2 messages (10 sec / 5 sec) * 5 MDB instances. In order to configure the number of MDB instances, open the JBoss configuration file (standalone.xml) and scroll down until the section regarding the <subsystem xmlns=”urn:jboss:domain:ejb3:1.2″> subsystem.

..
            <mdb>
                <resource-adapter-ref resource-adapter-name="hornetq-ra"/>
                <bean-instance-pool-ref pool-name="mdb-strict-max-pool"/>
            </mdb>
            <pools>
                <bean-instance-pools>
                    <strict-max-pool name="slsb-strict-max-pool" max-pool-size="20" instance-acquisition-timeout="5" instance-acquisition-timeout-unit="MINUTES"/>
                    <strict-max-pool name="mdb-strict-max-pool" max-pool-size="20" instance-acquisition-timeout="5" instance-acquisition-timeout-unit="MINUTES"/>
                </bean-instance-pools>
            </pools>
..

Not that the line <bean-instance-pool-ref pool-name=”mdb-strict-max-pool”/> links with the mdb-strict-max-pool pool, which in turn has its own configuration:

<strict-max-pool name="mdb-strict-max-pool" max-pool-size="20" instance-acquisition-timeout="5" instance-acquisition-timeout-unit="MINUTES"/>

As we can see, the default value is 20 instances for each MDB. When necessary, you just need to adjust the pool size.

Conclusion

Making use of JMS with JEE6 is quite a simple process, which requires a few annotations some lines of code. However, it’s a very robust and useful mechanism, extensively used in JEE applications.

The code can be found on GitHub and feel free to ask if you have any question.