Tech and Tips

ActiveMQ being a messaging system based on queues (aka FIFOs), one would take for granted that if there is only one producer and one consumer for a given queue (and they are both single threaded), the order of the messages is preserved.
Well, not always!

Let's say I have the following configuration for my ActiveMQ client running Camel:

<?xml version="1.0" encoding="UTF-8"?>
<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0">

    <bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="${env.activemq.broker.url}"/>
        <property name="userName" value="${env.activemq.broker.username}"/>
        <property name="password" value="${env.activemq.broker.password}"/>
    </bean>

    <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" init-method="start" destroy-method="stop">
        <property name="maxConnections" value="12"/>
        <property name="maximumActiveSessionPerConnection" value="200"/>
        <property name="connectionFactory" ref="amqConnectionFactory"/>
    </bean>

    <bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
        <property name="connectionFactory" ref="pooledConnectionFactory"/>
        <property name="transacted" value="true"/>
        <property name="cacheLevelName" value="CACHE_NONE"/>
    </bean>

    <bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
        <property name="configuration" ref="jmsConfig"/>
    </bean>

</blueprint>

With this configuration, even with a prefetch of 1 and only one consumer, you risk having messages being consumed out of order even if they were produced in the right order.
The culprit is CACHE_NONE, which you want to use if you are using XA transactions.
But in normal circumstances, with a local transaction manager or with the one built-in with the JmsConfiguration bean, it is recommended to use CACHE_CONSUMER not only to improve performance but also to ensure proper message ordering.

Side note regarding prefetch=1:
Even though one could expect having only one message sent to the consumer until it gets ack-ed (which is when you are done processing it when you have transacted=true), it is still possible to have a second message assigned to that consumer in the dispatch queue, which in effect get blocked until the first message is fully processed (which can be a problem for slow consumers).
The solution (if this is really a problem) would be to use prefetch=0 for that given consumer, but this is costly since the consumer is now polling the broker!

More info here


If message ordering is a big requirement for you, you might want to look at the Camel resequencer.


Update 20160624: And now there is a Jira for this!
[ENTMQ-1783] Combination of CACHE_NONE and Transacted Affects Message Ordering - JBoss Issue Tracker

I have been looking for a nice little ActiveMQ CLI utility in order to push/consume messages to/from ActiveMQ via the standard OpenWire protocol for quite a while now and I finally found it!

Thanks for the glorious developer behind this github project :)

ActiveMQ Command-line utility
http://nordlander.co/activemq-command-line-utility-a/

Make sure you have a look at all the parameters on the main github page of the project (https://github.com/fmtn/a) and use the jar with dependencies if you don't have maven installed.

Here is a little bash script you can use so make things simpler:

#!/bin/sh
#
# blog: http://nordlander.co/activemq-command-line-utility-a/
# github: https://github.com/fmtn/a
#
# Parameters examples:
# -U admin -P admin -p "toto" q_fake
# -U admin -P admin -c 9 -p "toto" q_fake
# -U admin -P admin -o /tmp/msgs/fakemsg -c 9 -g q_fake
# -U admin -P admin -p @/tmp/msgs/fakemsg-no4 q_fake
#
java -jar ~/bin/a-1.3.0-jar-with-dependencies.jar "$@"

Now if you have saved the script above under the file called amq_client.sh, you can use it like this:
$ ./amq_client.sh -U admin -P admin -p "toto" q_fake

Here is a lesson learnt regarding using ActiveMQ with several consumers for one queue and transactions enabled.

What we wanted to achieve was to have the following setup for each queue:

• several consumers to spread the load and avoid that one message gets stuck if a route gets into trouble or takes a long time to complete
• each consumer should get only one message at a time so no message gets stuck in the local client buffer
• the camel route should be transacted so if something goes wrong the message is not lost but gets delivered to another consumer

We are using camel and the camel-activemq component.
So defining several consumers is rather easy.
We used prefetch=1 to make sure that only one message at a time is delivered to a route .
And we used a transaction manager for our JmsConfiguration bean.
So my config looked like something like that

<bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
        <property name="connectionFactory" ref="pooledConnectionFactory" />
        <property name="transacted" value="true" /> 
        <property name="transactionManager" ref="jmsTransactionManager" />
        <property name="cacheLevelName" value="CACHE_CONSUMER" />
</bean>

So far nothing special.

What I then noticed was that sometimes a consumer could get 2 messages assigned, which puzzled me since I specified prefetch=1!
One being processed and waiting for an ACK to finish the transaction, and the other one just waiting to be taken and not being taken by another consumer even if one is available.
And this was bad because sometimes my route took several minutes to completed, effectively preventing the 2nd message to get through right away.

The culprit was CACHE_CONSUMER. This pre-assigns the next message to the consumer to speed things up.

So I changed this to CACHE_NONE and now everything behaves as expected!

Get the Hawtio app from http://hawt.io
Start it like this:

java -jar -Dhawtio.offline=true -Dhawtio.proxyWhitelist=* hawtio-app-2.0.2.jar

Then go to http://localhost:8080/hawtio (actually the tab should open by itself in your default browser after you ran the command above)




Choose Connect->Remote, then Add Connection and use port 8161 and path /api/jolokia




And voila :)


Remember to adjust 2 settings by clicking on the little user icon in the top right corner (in Preferences):
- the ActiveMQ username/password in order to send/consume messages
- the max collection size for Jolokia in order to get all the objects back from JMX

From ActiveMQ 5.9.1 and up, the jolokia agent is installed by default so you should not need to do anything special with your ActiveMQ installation.

Hawtio is not included by default in ActiveMQ 5.9.1 as it was in 5.9.0.

But by following the instructions from these 2 pages, it is fairly simple to add it manually:
http://sensatic.net/activemq/activemq-and-hawtio.html
http://activemq.2283324.n4.nabble.com/Hawto-log-in-td4673552.html

The main info from the second page is this missing info from the first one:
-Djava.security.auth.login.config=$ACTIVEMQ_CONF/login.config

Also make sure that you update the usernames/password in the following files in $ACTIVEMQ_HOME/conf:
- credentials.properties
- users.properties
- groups.properties
- login.config

Make sure you also add this to $ACTIVEMQ_HOME/bin/activemq, somewhere around line 160:
ACTIVEMQ_OPTS="$ACTIVEMQ_OPTS -Dhawtio.realm=activemq -Dhawtio.role=admins -Dhawtio.rolePrincipalClasses=org.apache.activemq.jaas.GroupPrincipal -Djava.security.auth.login.config=$ACTIVEMQ_CONF/login.config"
ACTIVEMQ_OPTS="$ACTIVEMQ_OPTS -Dhawtio.offline=true -Dhawtio.config.cloneOnStartup=false -Dhawtio.config.pullOnStartup=false"

The second set of sysetm properties is to prevent hawtio to try to fetch data from github which is does by default. If you are behind a firewall, this can cause hawtio to hang for over 2 minutes.

2 more things:
- if you have some special proxy settings, you might want to use 127.0.1.1 instead of 127.0.0.1 to reach your local hawtio
- the hawtio tree structure on the left side gets all screwed up each time a refresh occurs. I am not sure if this is a bug in hawtio but it is very annoying.

More links:
http://hawt.io/getstarted/
http://hawt.io/configuration/index.html#Configuration_Properties


Good luck!

With JMS persistent messages, all the message are persisted to the storage device by default, but also kept in memory. When a queue can potentially contain many messages, it is then necessary to set a max memory allowed, like this:

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="/var/activemq/data">
    <destinationPolicy>
        <policyMap>
            <policyEntries>
                <policyEntry queue=">" producerFlowControl="false" memoryLimit="100mb" queuePrefetch=”1” maxPageSize=”10” lazyDispatch=”true” />
            </policyEntries>
        </policyMap>
    </destinationPolicy>
</broker>

There is a way to add/remove queues on ActiveMQ from the command line.
It uses the REST interface of Jolokia which comes with Hawtio on ActiveMQ 5.9.0.

The first 2 commands contain all you need to create and then to remove the queue called q_titi, using the user admin/admin:
curl -u admin:admin -d "{\"type\":\"exec\",\"mbean\":\"org.apache.activemq:type=Broker,brokerName=localhost\",\"operation\":\"addQueue(java.lang.String)\",\"arguments\":[\"q_titi\"]}" http://localhost:8161/hawtio/jolokia/

curl -u admin:admin -d "{\"type\":\"exec\",\"mbean\":\"org.apache.activemq:type=Broker,brokerName=localhost\",\"operation\":\"removeQueue(java.lang.String)\",\"arguments\":[\"q_titi\"]}" http://localhost:8161/hawtio/jolokia/


The next 2 command do the same for the queue q_OMG but use the attached files for the necessary JSON request (so no more need to escape the double quotes and way more readable!):

curl -u admin:admin -d "@/tmp/jmr/createQ_json.txt" http://localhost:8161/hawtio/jolokia/
curl -u admin:admin -d "@/tmp/jmr/removeQ_json.txt" http://localhost:8161/hawtio/jolokia/

createQ_json.txt looks like this:

{
 "type":"exec",
 "mbean":"org.apache.activemq:type=Broker,brokerName=localhost",
 "operation":"addQueue(java.lang.String)",
 "arguments":["q_OMG"]
}

removeQ_json.txt looks like this:

{
 "type":"exec",
 "mbean":"org.apache.activemq:type=Broker,brokerName=localhost",
 "operation":"removeQueue(java.lang.String)",
 "arguments":["q_OMG"]
}

More info here:
Monitoring ActiveMQ via HTTP http://www.jakubkorab.net/2013/11/monitoring-activemq-via-http.html