This example demonstrates how to ensure strict ordering across a cluster using clustered message grouping
We create 3 nodes each with a grouping message handler, one with a Local handler and 2 with a Remote handler.
The local handler acts as an arbitrator for the 2 remote handlers, holding the information on routes and communicating the routing info with the remote handlers on the other 2 nodes
We then send some messages to each node with the same group id set and ensure the same consumer receives all of them
Here's the relevant snippet from the server configuration that has the local handler
<cluster-connections>
<cluster-connection name="my-cluster">
<address>jms</address>
<connector-ref>netty-connector</connector-ref>
<retry-interval>500</retry-interval>
<use-duplicate-detection>true</use-duplicate-detection>
<forward-when-no-consumers>true</forward-when-no-consumers>
<max-hops>1</max-hops>
<discovery-group-ref discovery-group-name="my-discovery-group"/>
</cluster-connection>
</cluster-connections>
<grouping-handler name="my-grouping-handler">
<type>LOCAL</type>
<address>jms</address>
<timeout>5000</timeout>
</grouping-handler>
Here's the relevant snippet from the server configuration that has the remote handlers
<cluster-connections>
<cluster-connection name="my-cluster">
<address>jms</address>
<retry-interval>500</retry-interval>
<use-duplicate-detection>true</use-duplicate-detection>
<forward-when-no-consumers>true</forward-when-no-consumers>
<max-hops>1</max-hops>
<discovery-group-ref discovery-group-name="my-discovery-group"/>
</cluster-connection>
</cluster-connections>
<grouping-handler name="my-grouping-handler">
<type>REMOTE</type>
<address>jms</address>
<timeout>5000</timeout>
</grouping-handler>
For more information on HornetQ clustering and grouping see the clustering and grouping section of the user manual.
To run the example, simply type mvn verify
from this directory
ic0 = getContext(0);
Queue queue = (Queue)ic0.lookup("/queue/exampleQueue");
ConnectionFactory cf0 = (ConnectionFactory)ic0.lookup("/ConnectionFactory");
ic1 = getContext(1);
ConnectionFactory cf1 = (ConnectionFactory)ic1.lookup("/ConnectionFactory");
ic2 = getContext(2);
ConnectionFactory cf2 = (ConnectionFactory)ic2.lookup("/ConnectionFactory");
connection0 = cf0.createConnection();
connection1 = cf1.createConnection();
connection2 = cf2.createConnection();
Session session0 = connection0.createSession(false, Session.AUTO_ACKNOWLEDGE);
Session session1 = connection1.createSession(false, Session.AUTO_ACKNOWLEDGE);
Session session2 = connection2.createSession(false, Session.AUTO_ACKNOWLEDGE);
connection0.start();
connection1.start();
connection2.start();
MessageConsumer consumer = session0.createConsumer(queue);
MessageProducer producer0 = session0.createProducer(queue);
MessageProducer producer1 = session1.createProducer(queue);
MessageProducer producer2 = session2.createProducer(queue);
final int numMessages = 10;
for (int i = 0; i < numMessages; i++)
{
TextMessage message = session0.createTextMessage("This is text message " + i);
message.setStringProperty(HornetQMessage.JMSXGROUPID, "Group-0");
producer0.send(message);
System.out.println("Sent messages: " + message.getText() + " to node 0");
}
for (int i = 0; i < numMessages; i++)
{
TextMessage message = session1.createTextMessage("This is text message " + (i + 10));
message.setStringProperty(HornetQMessage.JMSXGROUPID, "Group-0");
producer1.send(message);
System.out.println("Sent messages: " + message.getText() + " to node 1");
}
for (int i = 0; i < numMessages; i++)
{
TextMessage message = session2.createTextMessage("This is text message " + (i + 20));
message.setStringProperty(HornetQMessage.JMSXGROUPID, "Group-0");
producer2.send(message);
System.out.println("Sent messages: " + message.getText() + " to node 2");
}
for (int i = 0; i < numMessages * 3; i++)
{
TextMessage message0 = (TextMessage)consumer.receive(5000);
System.out.println("Got message: " + message0.getText() + " from node 0");
}
if (connection0 != null)
{
connection0.close();
}
if (connection1 != null)
{
connection1.close();
}
if (connection2 != null)
{
connection2.close();
}
if (ic0 != null)
{
ic0.close();
}
if (ic1 != null)
{
ic1.close();
}
if (ic2 != null)
{
ic2.close();
}