The HA-JMS client is different from regular JMS clients in two important aspects.
The HA-JMS client must look up JMS connection factories as well as queues and topicsusing HA-JNDI (the default port is 1100). This ensures the factory/queue/topic can be found no matter which cluster node is running the HA-JMS server.
If the client is a J2EE component (session bean or web application) running inside the AS, the lookup via HA-JNDI can be configured using the component's deployment descriptors: In the standard deployment descriptor (ejb-jar.xml or web.xml):
<resource-ref> <res-ref-name>jms/ConnectionFactory</res-ref-name> <res-type>javax.jms.QueueConnectionFactory</res-type> <res-auth>Container</res-auth> </resource-ref> <resource-ref> <res-ref-name>jms/Queue</res-ref-name> <res-type>javax.jms.Queue</res-type> <res-auth>Container</res-auth> </resource-ref>
And in the JBoss-specific descriptor (jboss.xml or jboss-web.xml):
<resource-ref> <res-ref-name>jms/ConnectionFactory</res-ref-name> <!-- Use the JMS Resource Adapter, let it deal with knowing where the JMS server is --> <jndi-name>java:/JmsXA</jndi-name> </resource-ref> <resource-ref> <res-ref-name>jms/Queue</res-ref-name> <!-- Use HA-JNDI so we can find the queue on any node --> <jndi-name>jnp://localhost:1100/queue/A</jndi-name> </resource-ref>
The HA-JMS client must deal with exceptions that will occur on the JMS connection if server failover occurs. Unlike, for example, clustered EJB proxies, the JMS connection object does not include automatic failover logic. If the HA-JMS service fails over to a different master node, all client operations on the current connection will fail with a JMSException. To deal with this:
If the client is running inside the application server, the client should obtain the ConnectionFactory by looking up java:/JmsXAin JNDI. This will find the JBoss JMS Resource Adapter; the resource adapter will handle the task of detecting server failover and reconnecting to the new server when it starts.
For clients outside the application server, the best approach is to register an ExceptionListener with the connection; the listener will get a callback if there is an exception on the connection. The callback should then handle the task of closing the old connection and reconnecting. Following is a example application that continuously sends messages to a queue, handling any exceptions that occur:
package com.test.hajms.client;
import javax.naming.InitialContext;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.Connection;
import javax.jms.Session;
import javax.jms.MessageProducer;
import javax.jms.Message;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.DeliveryMode;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
public class FailoverJMSClient
{
private static final Log log = LogFactory.getLog(FailoverJMSClient.class);
public static final int NUM_RETRIES = 3;
volatile boolean doSend = true;
ConnectionFactory connectionFactory;
Destination queue;
Connection connection;
Session session;
MessageProducer producer;
public static void main(String[] args) throws Exception
{
FailoverJMSClient jmsClient = new FailoverJMSClient();
jmsClient.setUpJMS();
jmsClient.sendMessages();
}
public boolean setUpJMS()
{
InitialContext ic;
try
{
// assume jndi.properties is configured for HA-JNDI
ic = new InitialContext();
connectionFactory = (ConnectionFactory)ic.lookup("ConnectionFactory");
queue = (Destination)ic.lookup("queue/FailoverTestQueue");
connection = connectionFactory.createConnection();
try
{
log.debug("Connection created ...");
// KEY - register for exception callbacks
connection.setExceptionListener(new ExceptionListenerImpl());
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
log.debug("Session created ...");
producer = session.createProducer(queue);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
log.debug("Producer created ...");
return true;
}
catch (Exception e)
{
// We failed so close the connection
try
{
connection.close();
}
catch (JMSException ignored)
{
// Pointless
}
// Rethrow the initial problem to where we will log it
throw e;
}
finally
{
// And close the initial context
// We don't want to wait for the garbage collector to close it
// otherwise we'll have useless hanging network connections
ic.close();
}
}
catch (Exception e)
{
log.error("Error setting up JMS", e);
return false;
}
}
public void sendMessages()
{
int cnt = 0;
while(doSend)
{
try
{
Thread.sleep(100);
Message m = session.createObjectMessage(new Integer(cnt++));
producer.send(m);
log.trace("message " + cnt + " sent");
}
catch(Exception e)
{
cnt--;
log.error(e.getMessage());
}
}
}
private class ExceptionListenerImpl implements ExceptionListener
{
public void onException(JMSException e)
{
for(int i = 0; i < NUM_RETRIES; i++)
{
log.warn("Connection has problems, trying to re-create it, attempt " +
(i + 1) + " ...");
try
{
connection.close(); // unregisters the ExceptionListener
}
catch(Exception e2) {
// I will get an Exception anyway, since the connection to the
//server is broken, but close() frees up resources associated
// with the connection
}
boolean setupOK = setUpJMS();
if (setupOK)
{
log.info("Connection re-established");
return;
}
else
{
log.warn("Re-creating connection failed, retrying ...");
}
}
log.error("Cannot re-establish connection, giving up ...");
doSend = false;
}
}
}