HornetQ User Manual

The software can be download from the Download page:http://hornetq.org/downloads.html

Red Hat kindly employs developers to work full time on HornetQ, they are:

And many thanks to all our contributors, both old and new who helped create HornetQ, for a full list of the people who made it happen, take a look at our team page.

Messaging systems allow you to loosely couple heterogeneous systems together, whilst typically providing reliability, transactions and many other features.

Unlike systems based on a Remote Procedure Call (RPC) pattern, messaging systems primarily use an asynchronous message passing pattern with no tight relationship between requests and responses. Most messaging systems also support a request-response mode but this is not a primary feature of messaging systems.

Designing systems to be asynchronous from end-to-end allows you to really take advantage of your hardware resources, minimizing the amount of threads blocking on IO operations, and to use your network bandwidth to its full capacity. With an RPC approach you have to wait for a response for each request you make so are limited by the network round trip time, or latency of your network. With an asynchronous system you can pipeline flows of messages in different directions, so are limited by the network bandwidth not the latency. This typically allows you to create much higher performance applications.

Messaging systems decouple the senders of messages from the consumers of messages. The senders and consumers of messages are completely independent and know nothing of each other. This allows you to create flexible, loosely coupled systems.

Often, large enterprises use a messaging system to implement a message bus which loosely couples heterogeneous systems together. Message buses often form the core of an Enterprise Service Bus. (ESB). Using a message bus to de-couple disparate systems can allow the system to grow and adapt more easily. It also allows more flexibility to add new systems or retire old ones since they don't have brittle dependencies on each other.

Messaging systems normally support two main styles of asynchronous messaging: message queue messaging (also known as point-to-point messaging) and publish subscribe messaging. We'll summarise them briefly here:

A key feature of most messaging systems is reliable messaging. With reliable messaging the server gives a guarantee that the message will be delivered once and only once to each consumer of a queue or each durable subscription of a topic, even in the event of system failure. This is crucial for many businesses; e.g. you don't want your orders fulfilled more than once or any of your orders to be lost.

In other cases you may not care about a once and only once delivery guarantee and are happy to cope with duplicate deliveries or lost messages - an example of this might be transient stock price updates - which are quickly superseded by the next update on the same stock. The messaging system allows you to configure which delivery guarantees you require.

Messaging systems typically support the sending and acknowledgement of multiple messages in a single local transaction. HornetQ also supports the sending and acknowledgement of message as part of a large global transaction - using the Java mapping of XA; JTA.

Messages are either durable or non durable. Durable messages will be persisted in permanent storage and will survive server failure or restart. Non durable messages will not survive server failure or restart. Examples of durable messages might be orders or trades, where they cannot be lost. An example of a non durable message might be a stock price update which is transitory and doesn't need to survive a restart.

How do client applications interact with messaging systems in order to send and consume messages?

Several messaging systems provide their own proprietary APIs with which the client communicates with the messaging system.

There are also some standard ways of operating with messaging systems and some emerging standards in this space.

Let's take a brief look at these:

High Availability (HA) means that the system should remain operational after failure of one or more of the servers. The degree of support for HA varies between various messaging systems.

HornetQ provides automatic failover where your sessions are automatically reconnected to the backup server on event of live server failure.

For more information on HA, please see Chapter 39, High Availability and Failover.

Many messaging systems allow you to create groups of messaging servers called clusters. Clusters allow the load of sending and consuming messages to be spread over many servers. This allows your system to scale horizontally by adding new servers to the cluster.

Degrees of support for clusters varies between messaging systems, with some systems having fairly basic clusters with the cluster members being hardly aware of each other.

HornetQ provides very configurable state-of-the-art clustering model where messages can be intelligently load balanced between the servers in the cluster, according to the number of consumers on each node, and whether they are ready for messages.

HornetQ also has the ability to automatically redistribute messages between nodes of a cluster to prevent starvation on any particular node.

For full details on clustering, please see Chapter 38, Clusters.

Some messaging systems allow isolated clusters or single nodes to be bridged together, typically over unreliable connections like a wide area network (WAN), or the internet.

A bridge normally consumes from a queue on one server and forwards messages to another queue on a different server. Bridges cope with unreliable connections, automatically reconnecting when the connections becomes available again.

HornetQ bridges can be configured with filter expressions to only forward certain messages, and transformation can also be hooked in.

HornetQ also allows routing between queues to be configured in server side configuration. This allows complex routing networks to be set up forwarding or copying messages from one destination to another, forming a global network of interconnected brokers.

For more information please see Chapter 36, Core Bridges and Chapter 35, Diverting and Splitting Message Flows.

HornetQ core is designed simply as set of Plain Old Java Objects (POJOs) - we hope you like its clean-cut design.

We've also designed it to have as few dependencies on external jars as possible. In fact, HornetQ core has only one jar dependency, netty.jar, other than the standard JDK classes! This is because we use some of the netty buffer classes internally.

This allows HornetQ to be easily embedded in your own project, or instantiated in any dependency injection framework such as JBoss Microcontainer, Spring or Google Guice.

Each HornetQ server has its own ultra high performance persistent journal, which it uses for message and other persistence.

Using a high performance journal allows outrageous persistence message performance, something not achievable when using a relational database for persistence.

HornetQ clients, potentially on different physical machines interact with the HornetQ server. HornetQ currently provides two APIs for messaging at the client side:

JMS semantics are implemented by a thin JMS facade layer on the client side.

The HornetQ server does not speak JMS and in fact does not know anything about JMS, it is a protocol agnostic messaging server designed to be used with multiple different protocols.

When a user uses the JMS API on the client side, all JMS interactions are translated into operations on the HornetQ core client API before being transferred over the wire using the HornetQ wire format.

The server always just deals with core API interactions.

A schematic illustrating this relationship is shown in figure 3.1 below:

Figure 3.1 shows two user applications interacting with a HornetQ server. User Application 1 is using the JMS API, while User Application 2 is using the core client API directly.

You can see from the diagram that the JMS API is implemented by a thin facade layer on the client side.

HornetQ core is designed as a set of simple POJOs so if you have an application that requires messaging functionality internally but you don't want to expose that as a HornetQ server you can directly instantiate and embed HornetQ servers in your own application.

For more information on embedding HornetQ, see Chapter 44, Embedding HornetQ.

HornetQ provides its own fully functional Java Connector Architecture (JCA) adaptor which enables it to be integrated easily into any JEE compliant application server or servlet engine.

JEE application servers provide Message Driven Beans (MDBs), which are a special type of Enterprise Java Beans (EJBs) that can process messages from sources such as JMS systems or mail systems.

Probably the most common use of an MDB is to consume messages from a JMS messaging system.

According to the JEE specification, a JEE application server uses a JCA adapter to integrate with a JMS messaging system so it can consume messages for MDBs.

However, the JCA adapter is not only used by the JEE application server for consuming messages via MDBs, it is also used when sending message to the JMS messaging system e.g. from inside an EJB or servlet.

When integrating with a JMS messaging system from inside a JEE application server it is always recommended that this is done via a JCA adaptor. In fact, communicating with a JMS messaging system directly, without using JCA would be illegal according to the JEE specification.

The application server's JCA service provides extra functionality such as connection pooling and automatic transaction enlistment, which are desirable when using messaging, say, from inside an EJB. It is possible to talk to a JMS messaging system directly from an EJB, MDB or servlet without going through a JCA adapter, but this is not recommended since you will not be able to take advantage of the JCA features, such as caching of JMS sessions, which can result in poor performance.

Figure 3.2 below shows a JEE application server integrating with a HornetQ server via the HornetQ JCA adaptor. Note that all communication between EJB sessions or entity beans and Message Driven beans go through the adaptor and not directly to HornetQ.

The large arrow with the prohibited sign shows an EJB session bean talking directly to the HornetQ server. This is not recommended as you'll most likely end up creating a new connection and session every time you want to interact from the EJB, which is an anti-pattern.

For more information on using the JCA adaptor, please see Chapter 32, Application Server Integration and Java EE.

HornetQ can also be deployed as a stand-alone server. This means a fully independent messaging server not dependent on a JEE application server.

The standard stand-alone messaging server configuration comprises a core messaging server, a JMS service and a JNDI service.

The role of the JMS Service is to deploy any JMS Queue, Topic and ConnectionFactory instances from any server side hornetq-jms.xml configuration files. It also provides a simple management API for creating and destroying Queues, Topics and ConnectionFactory instances which can be accessed via JMX or the connection. It is a separate service to the HornetQ core server, since the core server is JMS agnostic. If you don't want to deploy any JMS Queue, Topic or ConnectionFactory instances via server side XML configuration and don't require a JMS management API on the server side then you can disable this service.

We also include a JNDI server since JNDI is a common requirement when using JMS to lookup Queues, Topics and ConnectionFactory instances. If you do not require JNDI then this service can also be disabled. HornetQ allows you to programmatically create JMS and core objects directly on the client side as opposed to looking them up from JNDI, so a JNDI server is not always a requirement.

The stand-alone server configuration uses JBoss Microcontainer to instantiate and enforce dependencies between the components. JBoss Microcontainer is a very lightweight POJO bootstrapper.

The stand-alone server architecture is shown in figure 3.3 below:

For more information on server configuration files see Section 49.1, “Server Configuration”. $

In the distribution you will find a directory called bin.

cd into that directory and you will find a Unix/Linux script called run.sh and a windows batch file called run.bat

To run on Unix/Linux type ./run.sh

To run on Windows type run.bat

These scripts are very simple and basically just set-up the classpath and some JVM parameters and start the JBoss Microcontainer. The Microcontainer is a light weight container used to deploy the HornetQ POJO's

To stop the server you will also find a Unix/Linux script stop.sh and a windows batch file stop.bat

To run on Unix/Linux type ./stop.sh

To run on Windows type stop.bat

Please note that HornetQ requires a Java 6 or later runtime to run.

Both the run and the stop scripts use the config under config/stand-alone/non-clustered by default. The configuration can be changed by running ./run.sh ../config/stand-alone/clustered or another config of your choosing. This is the same for the stop script and the windows bat files.

The run scripts run.sh and run.bat set some JVM settings for tuning running on Java 6 and choosing the garbage collection policy. We recommend using a parallel garbage collection algorithm to smooth out latency and minimise large GC pauses.

By default HornetQ runs in a maximum of 1GiB of RAM. To increase the memory settings change the -Xms and -Xmx memory settings as you would for any Java program.

If you wish to add any more JVM arguments or tune the existing ones, the run scripts are the place to do it.

HornetQ looks for its configuration files on the Java classpath.

The scripts run.sh and run.bat specify the classpath when calling Java to run the server.

In the distribution, the run scripts will add the non clustered configuration directory to the classpath. This is a directory which contains a set of configuration files for running the HornetQ server in a basic non-clustered configuration. In the distribution this directory is config/stand-alone/non-clustered/ from the root of the distribution.

The distribution contains several standard configuration sets for running:

You can of course create your own configuration and specify any configuration directory when running the run script.

Just make sure the directory is on the classpath and HornetQ will search there when starting up.

If you're using the Asynchronous IO Journal on Linux, you need to specify java.library.path as a property on your Java options. This is done automatically in the run.sh script.

If you don't specify java.library.path at your Java options then the JVM will use the environment variable LD_LIBRARY_PATH.

HornetQ can take a system property on the command line for configuring logging.

For more information on configuring logging, please see Chapter 42, Logging.

The configuration directory is specified on the classpath in the run scripts run.sh and run.bat This directory can contain the following files.

It is also possible to use system property substitution in all the configuration files. by replacing a value with the name of a system property. Here is an example of this with a connector configuration:

<connector name="netty">
   <factory-class>org.hornetq.core.remoting.impl.netty.NettyConnectorFactory</factory-class>
   <param key="host" value="${hornetq.remoting.netty.host:localhost}" type="String"/>
   <param key="port" value="${hornetq.remoting.netty.port:5445}" type="Integer"/>
</connector>

Here you can see we have replaced 2 values with system properties hornetq.remoting.netty.host and hornetq.remoting.netty.port. These values will be replaced by the value found in the system property if there is one, if not they default back to localhost or 5445 respectively. It is also possible to not supply a default. i.e. ${hornetq.remoting.netty.host}, however the system property must be supplied in that case.

The stand-alone server is basically a set of POJOs which are instantiated by the light weight JBoss Microcontainer engine.

Let's take a look at an example beans file from the stand-alone server:

<?xml version="1.0" encoding="UTF-8"?>

<deployment xmlns="urn:jboss:bean-deployer:2.0">

   <!-- MBean server -->
   <bean name="MBeanServer" class="javax.management.MBeanServer">
      <constructor factoryClass="java.lang.management.ManagementFactory"
                   factoryMethod="getPlatformMBeanServer"/>
   </bean>

   <!-- The core configuration -->
   <bean name="Configuration" class="org.hornetq.core.config.impl.FileConfiguration">
   </bean>

    <!-- The security manager -->
   <bean name="HornetQSecurityManager" class="org.hornetq.spi.core.security.HornetQSecurityManagerImpl">
      <start ignored="true"/>
      <stop ignored="true"/>
   </bean>

    <!-- The core server -->
   <bean name="HornetQServer" class="org.hornetq.core.server.impl.HornetQServerImpl">
      <constructor>
         <parameter>
            <inject bean="Configuration"/>
         </parameter>
         <parameter>
            <inject bean="MBeanServer"/>
         </parameter>
         <parameter>
            <inject bean="HornetQSecurityManager"/>
         </parameter>
      </constructor>
      <start ignored="true"/>
      <stop ignored="true"/>
   </bean>

    <!-- The Stand alone server that controls the jndi server-->
   <bean name="StandaloneServer" class="org.hornetq.jms.server.impl.StandaloneNamingServer">
      <constructor>
         <parameter>
            <inject bean="HornetQServer"/>
         </parameter>
      </constructor>
      <property name="port">${jnp.port:1099}</property>
      <property name="bindAddress">${jnp.host:localhost}</property>
      <property name="rmiPort">${jnp.rmiPort:1098}</property>
      <property name="rmiBindAddress">${jnp.host:localhost}</property>
   </bean>

   <!-- The JMS server -->
   <bean name="JMSServerManager" class="org.hornetq.jms.server.impl.JMSServerManagerImpl">
      <constructor>
         <parameter>
            <inject bean="HornetQServer"/>
         </parameter>
      </constructor>
   </bean>


</deployment>

We can see that, as well as the core HornetQ server, the stand-alone server instantiates various different POJOs, lets look at them in turn:

<?xml version="1.0" encoding="UTF-8"?>
<server>
   <mbean code="org.hornetq.service.HornetQFileConfigurationService"
      name="org.hornetq:service=HornetQFileConfigurationService">
   </mbean>

   <mbean code="org.hornetq.service.JBossASSecurityManagerService"
      name="org.hornetq:service=JBossASSecurityManagerService">
   </mbean>

   <mbean code="org.hornetq.service.HornetQStarterService" 
      name="org.hornetq:service=HornetQStarterService">
      <!--lets let the JMS Server start us-->
         <attribute name="Start">false</attribute>

      <depends optional-attribute-name="SecurityManagerService"
         proxy-type="attribute">org.hornetq:service=JBossASSecurityManagerService</depends>
      <depends optional-attribute-name="ConfigurationService"
         proxy-type="attribute">org.hornetq:service=HornetQFileConfigurationService</depends>
   </mbean>

   <mbean code="org.hornetq.service.HornetQJMSStarterService"
      name="org.hornetq:service=HornetQJMSStarterService">
      <depends optional-attribute-name="HornetQServer"
         proxy-type="attribute">org.hornetq:service=HornetQStarterService</depends>
   </mbean>
</server>

This jboss-service.xml configuration file is included inside the hornetq-service.sar on AS4 with embedded HornetQ. As you can see, on this configuration file we are starting various services:

The configuration for the HornetQ core server is contained in hornetq-configuration.xml. This is what the FileConfiguration bean uses to configure the messaging server.

There are many attributes which you can configure HornetQ. In most cases the defaults will do fine, in fact every attribute can be defaulted which means a file with a single empty configuration element is a valid configuration file. The different configuration will be explained throughout the manual or you can refer to the configuration reference here.

For this chapter we're going to use a very simple ordering system as our example. It is a somewhat contrived example because of its extreme simplicity, but it serves to demonstrate the very basics of setting up and using JMS.

We will have a single JMS Queue called OrderQueue, and we will have a single MessageProducer sending an order message to the queue and a single MessageConsumer consuming the order message from the queue.

The queue will be a durable queue, i.e. it will survive a server restart or crash. We also want to pre-deploy the queue, i.e. specify the queue in the server JMS configuration so it is created automatically without us having to explicitly create it from the client.

The file hornetq-jms.xml on the server classpath contains any JMS Queue, Topic and ConnectionFactory instances that we wish to create and make available to lookup via the JNDI.

A JMS ConnectionFactory object is used by the client to make connections to the server. It knows the location of the server it is connecting to, as well as many other configuration parameters. In most cases the defaults will be acceptable.

We'll deploy a single JMS Queue and a single JMS Connection Factory instance on the server for this example but there are no limits to the number of Queues, Topics and Connection Factory instances you can deploy from the file. Here's our configuration:

<configuration xmlns="urn:hornetq" 
   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   xsi:schemaLocation="urn:hornetq ../schemas/hornetq-jms.xsd ">
    
   <connection-factory name="ConnectionFactory">
      <connectors>
         <connector-ref connector-name="netty"/>
      </connectors>
      <entries>
         <entry name="ConnectionFactory"/>
      </entries>
   </connection-factory>

   <queue name="OrderQueue">
      <entry name="queues/OrderQueue"/>
   </queue>
</configuration>

We deploy one ConnectionFactory called ConnectionFactory and bind it in just one place in JNDI as given by the entry element. ConnectionFactory instances can be bound in many places in JNDI if you require.

The JMS API doc provides several connection factories for applications. HornetQ JMS users can choose to configure the types for their connection factories. Each connection factory has a signature attribute and a xa parameter, the combination of which determines the type of the factory. Attribute signature has three possible string values, i.e. generic, queue and topic; xa is a boolean type parameter. The following table gives their configuration values for different connection factory interfaces.

<configuration xmlns="urn:hornetq" 
   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   xsi:schemaLocation="urn:hornetq ../schemas/hornetq-jms.xsd ">

   <connection-factory name="ConnectionFactory" signature="queue">
      <xa>true</xa>
      <connectors>
         <connector-ref connector-name="netty"/>
      </connectors>
      <entries>
         <entry name="ConnectionFactory"/>
      </entries>
   </connection-factory>
</configuration>

When using JNDI from the client side you need to specify a set of JNDI properties which tell the JNDI client where to locate the JNDI server, amongst other things. These are often specified in a file called jndi.properties on the client classpath, or you can specify them directly when creating the JNDI initial context. A full JNDI tutorial is outside the scope of this document, please see the Sun JNDI tutorial for more information on how to use JNDI.

For talking to the JBoss JNDI Server, the jndi properties will look something like this:

java.naming.factory.initial=org.jnp.interfaces.NamingContextFactory
java.naming.provider.url=jnp://myhost:1099
java.naming.factory.url.pkgs=org.jboss.naming:org.jnp.interfaces

Where myhost is the hostname or IP address of the JNDI server. 1099 is the port used by the JNDI server and may vary depending on how you have configured your JNDI server.

In the default standalone configuration, JNDI server ports are configured in the file hornetq-beans.xml by setting properties on the JNDIServer bean:

<bean name="StandaloneServer" class="org.hornetq.jms.server.impl.StandaloneNamingServer">
   <constructor>
      <parameter>
         <inject bean="HornetQServer"/>
      </parameter>
   </constructor>
   <property name="port">${jnp.port:1099}</property>
   <property name="bindAddress">${jnp.host:localhost}</property>
   <property name="rmiPort">${jnp.rmiPort:1098}</property>
   <property name="rmiBindAddress">${jnp.host:localhost}</property>
</bean>

Here's the code for the example:

First we'll create a JNDI initial context from which to lookup our JMS objects:

InitialContect ic = new InitialContext();

Now we'll look up the connection factory:

ConnectionFactory cf = (ConnectionFactory)ic.lookup("/ConnectionFactory");

And look up the Queue:

Queue orderQueue = (Queue)ic.lookup("/queues/OrderQueue");

Next we create a JMS connection using the connection factory:

Connection connection = cf.createConnection();

And we create a non transacted JMS Session, with AUTO_ACKNOWLEDGE acknowledge mode:

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

We create a MessageProducer that will send orders to the queue:

MessageProducer producer = session.createProducer(orderQueue);

And we create a MessageConsumer which will consume orders from the queue:

MessageConsumer consumer = session.createConsumer(orderQueue);

We make sure we start the connection, or delivery won't occur on it:

connection.start();

We create a simple TextMessage and send it:

TextMessage message = session.createTextMessage("This is an order");
producer.send(message);

And we consume the message:

TextMessage receivedMessage = (TextMessage)consumer.receive();
System.out.println("Got order: " + receivedMessage.getText());

It is as simple as that. For a wide range of working JMS examples please see the examples directory in the distribution.

Although it is a very common JMS usage pattern to lookup JMS Administered Objects (that's JMS Queue, Topic and ConnectionFactory instances) from JNDI, in some cases a JNDI server is not available and you still want to use JMS, or you just think "Why do I need JNDI? Why can't I just instantiate these objects directly?"

With HornetQ you can do exactly that. HornetQ supports the direct instantiation of JMS Queue, Topic and ConnectionFactory instances, so you don't have to use JNDI at all.

For a full working example of direct instantiation please see the JMS examples in Chapter 11, Examples.

Here's our simple example, rewritten to not use JNDI at all:

We create the JMS ConnectionFactory object via the HornetQJMSClient Utility class, note we need to provide connection parameters and specify which transport we are using, for more information on connectors please see Chapter 16, Configuring the Transport.

              
TransportConfiguration transportConfiguration = new TransportConfiguration(NettyConnectorFactory.class.getName());
ConnectionFactory cf = HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF,transportConfiguration);

We also create the JMS Queue object via the HornetQJMSClient Utility class:

Queue orderQueue = HornetQJMSClient.createQueue("OrderQueue");

Next we create a JMS connection using the connection factory:

Connection connection = cf.createConnection();

And we create a non transacted JMS Session, with AUTO_ACKNOWLEDGE acknowledge mode:

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

We create a MessageProducer that will send orders to the queue:

MessageProducer producer = session.createProducer(orderQueue);

And we create a MessageConsumer which will consume orders from the queue:

MessageConsumer consumer = session.createConsumer(orderQueue);

We make sure we start the connection, or delivery won't occur on it:

connection.start();

We create a simple TextMessage and send it:

TextMessage message = session.createTextMessage("This is an order");
producer.send(message);

And we consume the message:

TextMessage receivedMessage = (TextMessage)consumer.receive();
System.out.println("Got order: " + receivedMessage.getText());

This represents the client id for a JMS client and is needed for creating durable subscriptions. It is possible to configure this on the connection factory and can be set via the client-id element. Any connection created by this connection factory will have this set as its client id.

When the JMS acknowledge mode is set to DUPS_OK it is possible to configure the consumer so that it sends acknowledgements in batches rather that one at a time, saving valuable bandwidth. This can be configured via the connection factory via the dups-ok-batch-size element and is set in bytes. The default is 1024 * 1024 bytes = 1 MiB.

When receiving messages in a transaction it is possible to configure the consumer to send acknowledgements in batches rather than individually saving valuable bandwidth. This can be configured on the connection factory via the transaction-batch-size element and is set in bytes. The default is 1024 * 1024.

Some of the core messaging concepts are similar to JMS concepts, but core messaging concepts differ in some ways. In general the core messaging API is simpler than the JMS API, since we remove distinctions between queues, topics and subscriptions. We'll discuss each of the major core messaging concepts in turn, but to see the API in detail, please consult the Javadoc.

Here's a very simple program using the core messaging API to send and receive a message:

ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(
                                           InVMConnectorFactory.class.getName()));

ClientSessionFactory factory =  locator.createClientSessionFactory();

ClientSession session = factory.createSession();

session.createQueue("example", "example", true);

ClientProducer producer = session.createProducer("example");

ClientMessage message = session.createMessage(true);

message.getBodyBuffer().writeString("Hello");

producer.send(message);

session.start();

ClientConsumer consumer = session.createConsumer("example");

ClientMessage msgReceived = consumer.receive();

System.out.println("message = " + msgReceived.getBodyBuffer().readString());

session.close();

If you are using just a pure HornetQ Core client (i.e. no JMS) then you need hornetq-core-client.jar and netty.jar on your client classpath.

If the client runs inside a Java 5 virtual machine, use instead hornetq-core-client-java5.jar.

If you are using JMS on the client side, then you will also need to include hornetq-jms-client.jar and jboss-jms-api.jar.

If the client runs inside a Java 5 virtual machine, include instead hornetq-jms-client-java5.jar.

If you are looking up JMS resources from the JNDI server co-located with the HornetQ standalone server, you will also need the jar jnp-client.jar jar on your client classpath as well as any other jars mentioned previously.

To run a JMS example, simply cd into the appropriate example directory and type ./build.sh (or build.bat if you are on Windows).

Here's a listing of the examples with a brief description.

To run a core example, simply cd into the appropriate example directory and type ant

Most of the Java EE examples can be run the following way. simply cd into the appropriate example directory and type mvn test. This will use Arquillian to run the Application Server and deploy the application. Note that you must have jboss AS 7 installed and the JBOSS_HOME environment variable set. Please refer to the examples documentation for further instructions.

The bindings journal is configured using the following attributes in hornetq-configuration.xml

The jms config shares its configuration with the bindings journal.

The message journal is configured using the following attributes in hornetq-configuration.xml

The Java NIO journal gives great performance, but If you are running HornetQ using Linux Kernel 2.6 or later, we highly recommend you use the AIO journal for the very best persistence performance.

It's not possible to use the AIO journal under other operating systems or earlier versions of the Linux kernel.

If you are running Linux kernel 2.6 or later and don't already have libaio installed, you can easily install it using the following steps:

Using yum, (e.g. on Fedora or Red Hat Enterprise Linux):

yum install libaio

Using aptitude, (e.g. on Ubuntu or Debian system):

apt-get install libaio

In some situations, zero persistence is sometimes required for a messaging system. Configuring HornetQ to perform zero persistence is straightforward. Simply set the parameter persistence-enabled in hornetq-configuration.xml to false.

Please note that if you set this parameter to false, then zero persistence will occur. That means no bindings data, message data, large message data, duplicate id caches or paging data will be persisted.

You may want to inspect the existent records on each one of the journals used by HornetQ, and you can use the export/import tool for that purpose. The export/import are classes located at the hornetq-core.jar, you can export the journal as a text file by using this command:

java -cp hornetq-core.jar org.hornetq.core.journal.impl.ExportJournal <JournalDirectory> <JournalPrefix> <FileExtension> <FileSize> <FileOutput>

To import the file as binary data on the journal (Notice you also require netty.jar):

java -cp hornetq-core.jar:netty.jar org.hornetq.core.journal.impl.ImportJournal <JournalDirectory> <JournalPrefix> <FileExtension> <FileSize> <FileInput>

One of the most important concepts in HornetQ transports is the acceptor. Let's dive straight in and take a look at an acceptor defined in xml in the configuration file hornetq-configuration.xml.

<acceptors>
   <acceptor name="netty">
      <factory-class>
         org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory
      </factory-class>
      <param key="port" value="5446"/>
   </acceptor>
</acceptors>

Acceptors are always defined inside an acceptors element. There can be one or more acceptors defined in the acceptors element. There's no upper limit to the number of acceptors per server.

Each acceptor defines a way in which connections can be made to the HornetQ server.

In the above example we're defining an acceptor that uses Netty to listen for connections at port 5446.

The acceptor element contains a sub-element factory-class, this element defines the factory used to create acceptor instances. In this case we're using Netty to listen for connections so we use the Netty implementation of an AcceptorFactory to do this. Basically, the factory-class element determines which pluggable transport we're going to use to do the actual listening.

The acceptor element can also be configured with zero or more param sub-elements. Each param element defines a key-value pair. These key-value pairs are used to configure the specific transport, the set of valid key-value pairs depends on the specific transport be used and are passed straight through to the underlying transport.

Examples of key-value pairs for a particular transport would be, say, to configure the IP address to bind to, or the port to listen at.

Whereas acceptors are used on the server to define how we accept connections, connectors are used by a client to define how it connects to a server.

Let's look at a connector defined in our hornetq-configuration.xml file:

<connectors>
   <connector name="netty">
      <factory-class>
         org.hornetq.core.remoting.impl.netty.NettyConnectorFactory
      </factory-class>
      <param key="port" value="5446"/>
   </connector>
</connectors>

Connectors can be defined inside a connectors element. There can be one or more connectors defined in the connectors element. There's no upper limit to the number of connectors per server.

You make ask yourself, if connectors are used by the client to make connections then why are they defined on the server? There are a couple of reasons for this:

How do we configure a core ClientSessionFactory with the information that it needs to connect with a server?

Connectors are also used indirectly when directly configuring a core ClientSessionFactory to directly talk to a server. Although in this case there's no need to define such a connector in the server side configuration, instead we just create the parameters and tell the ClientSessionFactory which connector factory to use.

Here's an example of creating a ClientSessionFactory which will connect directly to the acceptor we defined earlier in this chapter, it uses the standard Netty TCP transport and will try and connect on port 5446 to localhost (default):

Map<String, Object> connectionParams = new HashMap<String, Object>();
    
connectionParams.put(org.hornetq.core.remoting.impl.netty.TransportConstants.PORT_PROP_NAME, 
                    5446);

TransportConfiguration transportConfiguration = 
    new TransportConfiguration(
    "org.hornetq.core.remoting.impl.netty.NettyConnectorFactory", 
    connectionParams);

ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(transportConfiguration);

ClientSessionFactory sessionFactory = locator.createClientSessionFactory();

ClientSession session = sessionFactory.createSession(...);

etc

Similarly, if you're using JMS, you can configure the JMS connection factory directly on the client side without having to define a connector on the server side or define a connection factory in hornetq-jms.xml:

Map<String, Object> connectionParams = new HashMap<String, Object>();

connectionParams.put(org.hornetq.core.remoting.impl.netty.TransportConstants.PORT_PROP_NAME, 5446);

TransportConfiguration transportConfiguration = 
    new TransportConfiguration(
    "org.hornetq.core.remoting.impl.netty.NettyConnectorFactory", 
    connectionParams);

ConnectionFactory connectionFactory = HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, transportConfiguration);

Connection jmsConnection = connectionFactory.createConnection();

etc

Out of the box, HornetQ currently uses Netty, a high performance low level network library.

Our Netty transport can be configured in several different ways; to use old (blocking) Java IO, or NIO (non-blocking), also to use straightforward TCP sockets, SSL, or to tunnel over HTTP or HTTPS, on top of that we also provide a servlet transport.

We believe this caters for the vast majority of transport requirements.

<connector name="netty-servlet">
   <factory-class>org.hornetq.core.remoting.impl.netty.NettyConnectorFactory</factory-class>
   <param key="host" value="localhost"/>
   <param key="port" value="8443"/>
   <param key="use-servlet" value="true"/>
   <param key="servlet-path" value="/messaging/HornetQServlet"/>
   <param key="ssl-enabled" value="true"/>
   <param key="key-store-path" value="path to a key-store"/>
   <param key="key-store-password" value="key-store password"/>
</connector>

<Connector protocol="HTTP/1.1" SSLEnabled="true"
     port="8443" address="${jboss.bind.address}"
     scheme="https" secure="true" clientAuth="false"
     keystoreFile="path to a keystore"
     keystorePass="keystore password" sslProtocol = "TLS" />

Before a HornetQ client application exits it is considered good practice that it should close its resources in a controlled manner, using a finally block.

Here's an example of a well behaved core client application closing its session and session factory in a finally block:

ServerLocator locator = null;
ClientSessionFactory sf = null;
ClientSession session = null;

try
{
   locator = HornetQClient.createServerLocatorWithoutHA(..);

   sf = locator.createClientSessionFactory();;

   session = sf.createSession(...);
   
   ... do some stuff with the session...
}
finally
{
   if (session != null)
   {
      session.close();
   }
   
   if (sf != null)
   {
      sf.close();
   }

   if(locator != null)
   {
      locator.close();
   }
}

And here's an example of a well behaved JMS client application:

Connection jmsConnection = null;

try
{
   ConnectionFactory jmsConnectionFactory = HornetQJMSClient.createConnectionFactoryWithoutHA(...);

   jmsConnection = jmsConnectionFactory.createConnection();

   ... do some stuff with the connection...
}
finally
{
   if (connection != null)
   {
      connection.close();
   }
}

Unfortunately users don't always write well behaved applications, and sometimes clients just crash so they don't have a chance to clean up their resources!

If this occurs then it can leave server side resources, like sessions, hanging on the server. If these were not removed they would cause a resource leak on the server and over time this result in the server running out of memory or other resources.

We have to balance the requirement for cleaning up dead client resources with the fact that sometimes the network between the client and the server can fail and then come back, allowing the client to reconnect. HornetQ supports client reconnection, so we don't want to clean up "dead" server side resources too soon or this will prevent any client from reconnecting, as it won't be able to find its old sessions on the server.

HornetQ makes all of this configurable. For each ClientSessionFactory we define a connection TTL. Basically, the TTL determines how long the server will keep a connection alive in the absence of any data arriving from the client. The client will automatically send "ping" packets periodically to prevent the server from closing it down. If the server doesn't receive any packets on a connection for the connection TTL time, then it will automatically close all the sessions on the server that relate to that connection.

If you're using JMS, the connection TTL is defined by the ConnectionTTL attribute on a HornetQConnectionFactory instance, or if you're deploying JMS connection factory instances direct into JNDI on the server side, you can specify it in the xml config, using the parameter connection-ttl.

The default value for connection ttl is 60000ms, i.e. 1 minute. A value of -1 for ConnectionTTL means the server will never time out the connection on the server side.

If you do not wish clients to be able to specify their own connection TTL, you can override all values used by a global value set on the server side. This can be done by specifying the connection-ttl-override attribute in the server side configuration. The default value for connection-ttl-override is -1 which means "do not override" (i.e. let clients use their own values).

[Finalizer] 20:14:43,244 WARNING [org.hornetq.core.client.impl.DelegatingSession]  I'm closing a ClientSession you left open. Please make sure you close all ClientSessions explicitly before let
ting them go out of scope!
[Finalizer] 20:14:43,244 WARNING [org.hornetq.core.client.impl.DelegatingSession]  The session you didn't close was created here:
java.lang.Exception
   at org.hornetq.core.client.impl.DelegatingSession.<init>(DelegatingSession.java:83)
   at org.acme.yourproject.YourClass (YourClass.java:666)

In the previous section we discussed how the client sends pings to the server and how "dead" connection resources are cleaned up by the server. There's also another reason for pinging, and that's for the client to be able to detect that the server or network has failed.

As long as the client is receiving data from the server it will consider the connection to be still alive.

If the client does not receive any packets for client-failure-check-period milliseconds then it will consider the connection failed and will either initiate failover, or call any FailureListener instances (or ExceptionListener instances if you are using JMS) depending on how it has been configured.

If you're using JMS it's defined by the ClientFailureCheckPeriod attribute on a HornetQConnectionFactory instance, or if you're deploying JMS connection factory instances direct into JNDI on the server side, you can specify it in the hornetq-jms.xml configuration file, using the parameter client-failure-check-period.

The default value for client failure check period is 30000ms, i.e. 30 seconds. A value of -1 means the client will never fail the connection on the client side if no data is received from the server. Typically this is much lower than connection TTL to allow clients to reconnect in case of transitory failure.

By default, packets received on the server side are executed on the remoting thread.

It is possible instead to use a thread from a thread pool to handle some packets so that the remoting thread is not tied up for too long. However, please note that processing operations asynchronously on another thread adds a little more latency. Please note that most short running operations are always handled on the remoting thread for performance reasons. To enable asynchronous connection execution, set the parameter async-connection-execution-enabled in hornetq-configuration.xml to true (default value is true).

This controls the flow of data between the server and the client as the client consumes messages. For performance reasons clients normally buffer messages before delivering to the consumer via the receive() method or asynchronously via a message listener. If the consumer cannot process messages as fast as they are being delivered and stored in the internal buffer, then you could end up with a situation where messages would keep building up possibly causing out of memory on the client if they cannot be processed in time.

<connection-factory name="ConnectionFactory">
   <connectors>
      <connector-ref connector-name="netty-connector"/>
   </connectors>
   <entries>
      <entry name="ConnectionFactory"/>
   </entries>
      
   <!-- Set the consumer window size to 0 to have *no* buffer on the client side -->
   <consumer-window-size>0</consumer-window-size>
</connection-factory>

<connection-factory name="ConnectionFactory">
   <connectors>
      <connector-ref connector-name="netty-connector"/>
   </connectors>
   <entries>
      <entry name="ConnectionFactory"/>
   </entries>
   <!-- We limit consumers created on this connection factory to consume messages at a maximum rate
   of 10 messages per sec -->
   <consumer-max-rate>10</consumer-max-rate>
</connection-factory>

HornetQ also can limit the amount of data sent from a client to a server to prevent the server being overwhelmed.

<connection-factory name="ConnectionFactory">
   <connectors>
      <connector-ref connector-name="netty-connector"/>
   </connectors>
   <entries>
      <entry name="ConnectionFactory"/>
   </entries>
   <producer-window-size>10</producer-window-size>
</connection-factory>

<address-settings>
   <address-setting match="jms.queue.exampleQueue">
      <max-size-bytes>100000</max-size-bytes>
      <address-full-policy>BLOCK</address-full-policy>
   </address-setting>
</address-settings>

<connection-factory name="ConnectionFactory">
   <connectors>
      <connector-ref connector-name="netty-connector"/>
   </connectors>
   <entries>
      <entry name="ConnectionFactory"/>
   </entries>
   <!-- We limit producers created on this connection factory to produce messages at a maximum rate
   of 10 messages per sec -->
   <producer-max-rate>10</producer-max-rate>
</connection-factory>

When committing or rolling back a transaction with HornetQ, the request to commit or rollback is sent to the server, and the call will block on the client side until a response has been received from the server that the commit or rollback was executed.

When the commit or rollback is received on the server, it will be committed to the journal, and depending on the value of the parameter journal-sync-transactional the server will ensure that the commit or rollback is durably persisted to storage before sending the response back to the client. If this parameter has the value false then commit or rollback may not actually get persisted to storage until some time after the response has been sent to the client. In event of server failure this may mean the commit or rollback never gets persisted to storage. The default value of this parameter is true so the client can be sure all transaction commits or rollbacks have been persisted to storage by the time the call to commit or rollback returns.

Setting this parameter to false can improve performance at the expense of some loss of transaction durability.

This parameter is set in hornetq-configuration.xml

If you are sending messages to a server using a non transacted session, HornetQ can be configured to block the call to send until the message has definitely reached the server, and a response has been sent back to the client. This can be configured individually for durable and non-durable messages, and is determined by the following two parameters:

Setting block on sends to true can reduce performance since each send requires a network round trip before the next send can be performed. This means the performance of sending messages will be limited by the network round trip time (RTT) of your network, rather than the bandwidth of your network. For better performance we recommend either batching many messages sends together in a transaction since with a transactional session, only the commit / rollback blocks not every send, or, using HornetQ's advanced asynchronous send acknowledgements feature described in Section 20.4, “Asynchronous Send Acknowledgements”.

If you are using JMS and you're using the JMS service on the server to load your JMS connection factory instances into JNDI then these parameters can be configured in hornetq-jms.xml using the elements block-on-durable-send and block-on-non-durable-send. If you're using JMS but not using JNDI then you can set these values directly on the HornetQConnectionFactory instance using the appropriate setter methods.

If you're using core you can set these values directly on the ClientSessionFactory instance using the appropriate setter methods.

When the server receives a message sent from a non transactional session, and that message is durable and the message is routed to at least one durable queue, then the server will persist the message in permanent storage. If the journal parameter journal-sync-non-transactional is set to true the server will not send a response back to the client until the message has been persisted and the server has a guarantee that the data has been persisted to disk. The default value for this parameter is true.

If you are acknowledging the delivery of a message at the client side using a non transacted session, HornetQ can be configured to block the call to acknowledge until the acknowledge has definitely reached the server, and a response has been sent back to the client. This is configured with the parameter BlockOnAcknowledge. If this is set to true then all calls to acknowledge on non transacted sessions will block until the acknowledge has reached the server, and a response has been sent back. You might want to set this to true if you want to implement a strict at most once delivery policy. The default value is false

If you are using a non transacted session but want a guarantee that every message sent to the server has reached it, then, as discussed in Section 20.2, “Guarantees of Non Transactional Message Sends”, you can configure HornetQ to block the call to send until the server has received the message, persisted it and sent back a response. This works well but has a severe performance penalty - each call to send needs to block for at least the time of a network round trip (RTT) - the performance of sending is thus limited by the latency of the network, not limited by the network bandwidth.

Let's do a little bit of maths to see how severe that is. We'll consider a standard 1Gib ethernet network with a network round trip between the server and the client of 0.25 ms.

With a RTT of 0.25 ms, the client can send at most 1000/ 0.25 = 4000 messages per second if it blocks on each message send.

If each message is < 1500 bytes and a standard 1500 bytes MTU size is used on the network, then a 1GiB network has a theoretical upper limit of (1024 * 1024 * 1024 / 8) / 1500 = 89478 messages per second if messages are sent without blocking! These figures aren't an exact science but you can clearly see that being limited by network RTT can have serious effect on performance.

To remedy this, HornetQ provides an advanced new feature called asynchronous send acknowledgements. With this feature, HornetQ can be configured to send messages without blocking in one direction and asynchronously getting acknowledgement from the server that the messages were received in a separate stream. By de-coupling the send from the acknowledgement of the send, the system is not limited by the network RTT, but is limited by the network bandwidth. Consequently better throughput can be achieved than is possible using a blocking approach, while at the same time having absolute guarantees that messages have successfully reached the server.

The window size for send acknowledgements is determined by the confirmation-window-size parameter on the connection factory or client session factory. Please see Chapter 34, Client Reconnection and Session Reattachment for more info on this.

Delaying redelivery can often be useful in the case that clients regularly fail or rollback. Without a delayed redelivery, the system can get into a "thrashing" state, with delivery being attempted, the client rolling back, and delivery being re-attempted ad infinitum in quick succession, consuming valuable CPU and network resources.

<!-- delay redelivery of messages for 5s -->
<address-setting match="jms.queue.exampleQueue">
   <redelivery-delay>5000</redelivery-delay>
</address-setting>

To prevent a client infinitely receiving the same undelivered message (regardless of what is causing the unsuccessful deliveries), messaging systems define dead letter addresses: after a specified unsuccessful delivery attempts, the message is removed from the queue and send instead to a dead letter address.

Any such messages can then be diverted to queue(s) where they can later be perused by the system administrator for action to be taken.

HornetQ's addresses can be assigned a dead letter address. Once the messages have be unsuccessfully delivered for a given number of attempts, they are removed from the queue and sent to the dead letter address. These dead letter messages can later be consumed for further inspection.

<!-- undelivered messages in exampleQueue will be sent to the dead letter address
   deadLetterQueue after 3 unsuccessful delivery attempts -->
<address-setting match="jms.queue.exampleQueue">
   <dead-letter-address>jms.queue.deadLetterQueue</dead-letter-address>
   <max-delivery-attempts>3</max-delivery-attempts>
</address-setting>

In normal use, HornetQ does not update delivery count persistently until a message is rolled back (i.e. the delivery count is not updated before the message is delivered to the consumer). In most messaging use cases, the messages are consumed, acknowledged and forgotten as soon as they are consumed. In these cases, updating the delivery count persistently before delivering the message would add an extra persistent step for each message delivered, implying a significant performance penalty.

However, if the delivery count is not updated persistently before the message delivery happens, in the event of a server crash, messages might have been delivered but that will not have been reflected in the delivery count. During the recovery phase, the server will not have knowledge of that and will deliver the message with redelivered set to false while it should be true.

As this behavior breaks strict JMS semantics, HornetQ allows to persist delivery count before message delivery but disabled it by default for performance implications.

To enable it, set persist-delivery-count-before-delivery to true in hornetq-configuration.xml:

<persist-delivery-count-before-delivery>true</persist-delivery-count-before-delivery>

Using HornetQ Core API, you can set an expiration time directly on the message:

// message will expire in 5000ms from now
message.setExpiration(System.currentTimeMillis() + 5000);

JMS MessageProducer allows to set a TimeToLive for the messages it sent:

// messages sent by this producer will be retained for 5s (5000ms) before expiration           
producer.setTimeToLive(5000);

Expired messages which are consumed from an expiry address have the following properties:

Expiry address are defined in the address-setting configuration:

<!-- expired messages in exampleQueue will be sent to the expiry address expiryQueue -->
<address-setting match="jms.queue.exampleQueue">
   <expiry-address>jms.queue.expiryQueue</expiry-address>
</address-setting>

If messages are expired and no expiry address is specified, messages are simply removed from the queue and dropped. Address wildcards can be used to configure expiry address for a set of addresses (see Chapter 13, Understanding the HornetQ Wildcard Syntax).

A reaper thread will periodically inspect the queues to check if messages have expired.

The reaper thread can be configured with the following properties in hornetq-configuration.xml

See Section 11.1.23, “Message Expiration” for an example which shows how message expiry is configured and used with JMS.

Large messages are stored on a disk directory on the server side, as configured on the main configuration file.

The configuration property large-messages-directory specifies where large messages are stored.

<configuration xmlns="urn:hornetq"
   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   xsi:schemaLocation="urn:hornetq /schema/hornetq-configuration.xsd">
...
<large-messages-directory>/data/large-messages</large-messages-directory>
...
</configuration

By default the large message directory is data/largemessages

For the best performance we recommend large messages directory is stored on a different physical volume to the message journal or paging directory.

Any message larger than a certain size is considered a large message. Large messages will be split up and sent in fragments. This is determined by the parameter min-large-message-size

The default value is 100KiB.

ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(NettyConnectorFactory.class.getName()))

locator.setMinLargeMessageSize(25 * 1024);

ClientSessionFactory factory = HornetQClient.createClientSessionFactory();

...
<connection-factory name="ConnectionFactory">
   <connectors>
      <connector-ref connector-name="netty"/>
   </connectors>
   <entries>
      <entry name="ConnectionFactory"/>
      <entry name="XAConnectionFactory"/>
   </entries>

   <min-large-message-size>250000</min-large-message-size>
</connection-factory>
...

...
<connection-factory name="ConnectionFactory">
   <connectors>
      <connector-ref connector-name="netty"/>
   </connectors>
...
   <compress-large-messages>true</compress-large-messages>
</connection-factory>
...

...
ClientMessage msg = consumer.receive(...);


// This will block here until the stream was transferred
msg.saveOutputStream(someOutputStream); 

ClientMessage msg2 = consumer.receive(...);

// This will not wait the transfer to finish
msg.setOutputStream(someOtherOutputStream); 
...

...
ClientMessage msg = session.createMessage();
msg.setInputStream(dataInputStream);
...

BytesMessage message = session.createBytesMessage();

FileInputStream fileInputStream = new FileInputStream(fileInput);

BufferedInputStream bufferedInput = new BufferedInputStream(fileInputStream);

message.setObjectProperty("JMS_HQ_InputStream", bufferedInput);

someProducer.send(message);

BytesMessage messageReceived = (BytesMessage)messageConsumer.receive(120000);
                
File outputFile = new File("huge_message_received.dat");
                
FileOutputStream fileOutputStream = new FileOutputStream(outputFile);
                
BufferedOutputStream bufferedOutput = new BufferedOutputStream(fileOutputStream);
                
// This will block until the entire content is saved on disk
messageReceived.setObjectProperty("JMS_HQ_SaveStream", bufferedOutput);

// This won't wait the stream to finish. You need to keep the consumer active.
messageReceived.setObjectProperty("JMS_HQ_OutputStream", bufferedOutput);

ClientMessage msg = consumer.receive();
         
byte[] bytes = new byte[1024];
for (int i = 0 ;  i < msg.getBodySize(); i += bytes.length)
{
   msg.getBody().readBytes(bytes);
   // Whatever you want to do with the bytes
}

BytesMessage rm = (BytesMessage)cons.receive(10000);

byte data[] = new byte[1024];

for (int i = 0; i < rm.getBodyLength(); i += 1024)
{
   int numberOfBytes = rm.readBytes(data);
   // Do whatever you want with the data
}