ActiveMQ 5.1.0 tutorial

It is always good to know at least one message brokers. ActiveMQ is one of those top brokers used actively industry wide. So, here is a small tutorial or tips to use ActiveMQ along with java messaging service.

Installations

I am using the following configuration on my system:

1. Windows XP
2. JDK 5 update 15
3. ActiveMQ 5.1.0

Download activeMQ from http://activemq.apache.org/. Unzip to any suitable location. And the installation is done!!!


Directory Structure

After you unzip, there are few basic files you need to know.

a. The bin folder contains the batch file, activemq.bat, using which you can start the server. It also contains activemq-admin.bat, using which you can get more details about activemq, like a listing of queues etc.

b. The conf folder contains the activemq.xml configuring the ActiveMQ broker. Here is where we can configure the transports, permanent queues etc.


Creating Temporary Queues

Firt run bin/activemq.bat. This should start ActiveMQ listening at port 61616. There is an admin console(http://localhost:8161/admin) that can be used to monitor the ActiveMQ server. It can also be used to create queues, but these are temporary queues. That means, once the server is shutdown, and restarted, the queues will have to be re created.

ActiveMQ is primarily build for creating queues dynamically. But, it is possible to create queues permanently in ActiveMQ 4.1 or above.


Creating Permanent Queues

The conf/activemq.xml is called broker configuration. By adding the below in that creates a permanent queue, i.e on restart of the server, the queue does exist. You dont have to create it again.

<beans>
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer" />
<broker xmlns="http://activemq.apache.org/schema/core">
<destinations>
<queue physicalName="test.activemq.queue" />
</destinations>
</broker>
</beans>


Creating Dynamic Queues and configuration

There are two ways to create queues, and configure ActiveMQ dynamically.

a. Programatically, in java code.
b. Using JNDI

The preferred approach is JNDI. But, there is nothing wrong in knowing the first one either.

a. Programmatic usage in ActiveMQ

// Create a ConnectionFactory
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");

// Create a Connection
Connection connection = connectionFactory.createConnection();
connection.start();

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

// Create the Queue
Destination destination = session.createQueue("test.prog.queue");

b. Using JNDI

Here, jndi.properties is used to configure. This should be present in the CLASSPATH for the queues, connection factory to be configured correctly.
This is taken from the activeMQ website and modified.

java.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFactory

# use the following property to configure the default connector
java.naming.provider.url = tcp://localhost:61616

# use the following property to specify the JNDI name the connection factory
# should appear as.
connectionFactoryNames = connectionFactory, queueConnectionFactory, topicConnectionFactry

# register some queues in JNDI using the form
# queue.[jndiName] = [physicalName]
queue.jndiqueue.test = test.prog.queue


# register some topics in JNDI using the form
# topic.[jndiName] = [physicalName]
#topic.MyTopic = example.MyTopic

Changing the ActiveMQ listening port

Copy and paste the conf/activemq.xml, and rename to newBroker.xml. change the port address in the XML:

<!-- The transport connectors ActiveMQ will listen to -->
<transportConnectors>
<transportConnector name="openwire" uri="tcp://localhost:51616" discoveryUri="multicast://default"/>
<transportConnector name="ssl" uri="ssl://localhost:51617"/>
<transportConnector name="stomp" uri="stomp://localhost:51613"/>
<transportConnector name="xmpp" uri="xmpp://localhost:51222"/>
</transportConnectors>

Run the activeMQ batch file with command:

activemq xbean:newBroker.xml

Or place the newBroker.xml in some other directory, say, c:\configs\activemq\newBroker.xml.

Run:

activemq xbean:file:c:/configs/activemq/newBroker.xml

which will start the broker at new ports.

Points to remember:

a. Below exception indicates you are using backslashes in the file: at commandline. change to forward slashes, it will resolve the exception.

ERROR: java.net.URISyntaxException: Illegal character in opaque part at index 13
: xbean:file:E:\soft\activemq\apache-activemq-5.1.0\conf\newBroker.xml
java.net.URISyntaxException: Illegal character in opaque part at index 13: xbean
:file:E:\soft\activemq\apache-activemq-5.1.0\conf\newBroker.xml
at java.net.URI$Parser.fail(URI.java:2816)
at java.net.URI$Parser.checkChars(URI.java:2989)

b. Keep the port number in check, a very large value will also result in a out of range exception.

Running two brokers on same machine

Other than changing the port as shown above, a new datastore has to be given. This is configured in the below XML snippet in the newBroker.xml.

<persistenceAdapter>
<amqPersistenceAdapter syncOnWrite="false" directory="${activemq.base}/newBrokerdata" maxFileLength="20 mb"/>
</persistenceAdapter>

Also, though not required, you can change the jetty servlet engine port of the newBroker.xml as below to say 8162.

<jetty xmlns="http://mortbay.com/schemas/jetty/1.0">
<connectors>
<nioConnector port="8162"/>
</connectors>
<handlers>
<webAppContext contextPath="/admin" resourceBase="${activemq.base}/webapps/admin" logUrlOnStart="true"/>
<webAppContext contextPath="/demo" resourceBase="${activemq.base}/webapps/demo" logUrlOnStart="true"/>
<webAppContext contextPath="/fileserver" resourceBase="${activemq.base}/webapps/fileserver" logUrlOnStart="true"/>
</handlers>
</jetty>

Network of Brokers

We can have a network of brokers for load balancing and support even if one of the brokers fail due to any reason, say, network problem.

The network of broker can be established in two ways:

a. Statically listing the ips of the other broker.
b. using auto discovery mode.

Both the configuration uses the networkConnector element in the broker xml configuration.

a. For statically listing the ips, the below xml can be used in our newBroker.xml

<!-- The store and forward broker networks ActiveMQ will listen to -->
<networkConnectors>
<!-- by default just auto discover the other brokers -->
<!-- <networkConnector name="default-nc" uri="multicast://default"/> -->
<!-- Example of a static configuration:
<networkConnector name="host1 and host2" uri="static://(tcp://host1:61616,tcp://host2:61616)"/>
-->
<networkConnector name="newBroker" uri="static://(tcp://localhost:61616)"/>
</networkConnectors>

If you observe the console, then the following line confirms that our brokers are running as a network of brokers.

INFO DemandForwardingBridge - Network connection between vm://localhost#0 and tcp://localhost/127.0.0.1:61616(localhost) has been established.

b. For auto discovrey mode, you can use the following XML.

<!-- The store and forward broker networks ActiveMQ will listen to -->
<networkConnectors>
<!-- by default just auto discover the other brokers -->
<networkConnector name="default-nc" uri="multicast://default"/>
<!-- Example of a static configuration:
<networkConnector name="host1 and host2" uri="static://(tcp://host1:61616,tcp://host2:61616)"/>
-->
</networkConnectors>

Yes, you guessed right, this is the default configuration.

Remenber to have the below xml snipet with discoveryUri attribute in all your brokers. By default, this is present.

<transportConnector name="openwire" uri="tcp://localhost:51616" discoveryUri="multicast://default"/>

Client using failover over a network of broker

The failover protocol has to be used on the client side, so that, if any of the broker in the network of brokers fail, then the client can use an alternative broker which is up and running.

This is also just a configuration.

For our example, this will be in jndi.properties as:

java.naming.provider.url = failover:(tcp://localhost:61616,tcp://localhost:51616)

Or in code as:
String localURI = "tcp://localhost:61616";
String remoteURI = "tcp://localhost:51616";
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:("+localURI+","+"remoteURI)");