techblog 

I’ve just started to look in to the various message queuing servers for our applications and first up is Sun’s Open Message Queue. I’m drawn to this as it is open source, has C and Java client APIs and also has high availability features using a persistent store.

Setting up a test environment was quite straight forward, as was using it from Spring. Here’s how I did it…

After unpacking the mq distribution it seems necessary to start the server once for it to generate a base set of files.

./imqbrokerd

Edit the configuration file to add properties for the JDBC datastore

var/instances/imqbroker/props/config.properties:

imq.persist.store=jdbc
imq.persist.jdbc.dbVendor=oracle
imq.persist.jdbc.oracle.driver=oracle.jdbc.pool.OracleConnectionPoolDataSource
imq.persist.jdbc.oracle.property.url=jdbc:oracle:thin:@testtom:1521:testtom
imq.persist.jdbc.oracle.user=tom
imq.persist.jdbc.oracle.needpassword=true
imq.persist.jdbc.oracle.password=secret

and the cluster

imq.cluster.ha=true
imq.cluster.brokerlist=bloo.xyz.uk,testtom
imq.cluster.clusterid=TomsCluster

Only one property for each broker needs to be unique

imq.brokerid=testtomBroker

Create the database tables for the data store (make sure your JDBC driver is in the classpath!)

./imqdbmgr create tbl

Start some/all of the nodes and then run the following on one of the nodes to check that the cluster is working

./imqcmd list bkr -u admin

-------------------------------------------------------------------------------------------------------------
ID of broker       Time since last
Broker ID             Address              State     Msgs in store   performing takeover   status timestamp
-------------------------------------------------------------------------------------------------------------
testtomBroker   testtom:7676               OPERATING   0                                     16 seconds
blooBroker      bloo:7676                  OPERATING   0                                     > 1 hour

A new message queue can then be set up with

./imqcmd create dst -u admin -t q -n testmq

    <bean id="mqConnectionFactoryInitialiser"
     class="uk.nominet.mq.MqConnectionFactoryInitialiser">
        <property name="configuration">
            <props>
                <prop key="imqAddressList">bloo,testtom</prop>
                <prop key="imqAddressListBehavior">RANDOM</prop>
            </props>
        </property>
    </bean>

The familiar Spring JMS configuration looks like

    <bean id="mqConnectionFactory"
     factory-bean="mqConnectionFactoryInitialiser"
     factory-method="createConnectionFactory"/>   

    <bean id="testmq" class="com.sun.messaging.Queue">
        <constructor-arg type="java.lang.String" value="testmq"/>
    </bean>   

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="mqConnectionFactory"/>
        <property name="defaultDestination" ref="testmq"/>
        <property name="receiveTimeout" value="20000"/>
    </bean>   

    <-- This is for sending –>
    <bean id=”activateContactMessageSender”
          class=”mq.example.ExampleMessageSender”>
        <property name=”jmsTemplate” ref=”jmsTemplate”/>
    </bean>   

    <– This is for receiving –>
    <bean id=”jmsContainer”
          class=”org.springframework.jms.listener.DefaultMessageListenerContainer”>
        <property name=”connectionFactory” ref=”mqConnectionFactory”/>
        <property name=”destination” ref=”testmq”/>
        <property name=”messageListener” ref=”messageListener1″ />
    </bean>   

    <bean id=”messageListener1″
              class=”org.springframework.jms.listener.adapter.MessageListenerAdapter“>
        <constructor-arg>
            <bean class=”mq.example.ExampleMessageListener”/>
        </constructor-arg>
    </bean>

Obviously the sender and receiver don’t have to be in the same application context (the two links provide good examples on how to implement the sender and receiver). Messages should be sent randomly to either broker and processed once by the receiver. If one of the brokers is taken down the messages will still get to the receiver via the other broker.

Update: Here’s the code for MqConnectionFactoryInitialiser

public class MqConnectionFactoryInitialiser { 
    private Properties configuration; 
 
    public void setConfiguration(Properties configuration) { 
        this.configuration = configuration; 
    } 
 
    public ConnectionFactory createConnectionFactory(){ 
        com.sun.messaging.ConnectionFactory cf = new com.sun.messaging.ConnectionFactory(); 
        try{ 
            for(Object key: configuration.keySet()){ 
                cf.setProperty((String) key, (String) configuration.get(key)); 
            } 
        } catch (JMSException e){ 
            throw new RuntimeException("Error configuring ConnectionFactory", e); 
        } 
        return cf; 
    } 
}