JMS Message Groups in Apache Camel

Message groups in JMS provide a way to identify a set of related messages. The messages could be related by anything - a customer order number, for example. Basically a JMS broker provides a guarantee that any messages that belong to a specific group will always be consumed by a common consumer. For instance, imagine that we’ve used the splitter pattern to split out line items from an order but want to aggregate those line items together later in a route. In order to perform that aggregation you need to guarantee that all of the messages being aggregated together are consumed by the same consumer.

Below is an example of using message groups with ActiveMQ within Apache Camel.

package com.brinksys.camel;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.camel.component.ActiveMQComponent;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;

import java.util.concurrent.TimeUnit;

public class App {
    private static BrokerService broker;

    public static void main(String[] args) throws Exception {
        try {
            startBroker();


            CamelContext ctx = createCamelContext();
            ctx.start();
            ctx.addRoutes(new RouteBuilder() {
                @Override
                public void configure() throws Exception {
                    /* Our direct route will take a message, and set the message to group 1 if the body is an integer,
                     * otherwise set the group to 2.
                     *
                     * This demonstrates the following concepts:
                     *  1) Header Manipulation
                     *  2) Checking the payload type of the body and using it in a choice.
                     *  3) JMS Message groups
                     */

                    from("direct:begin")
                    .choice()
                        .when(body().isInstanceOf(Integer.class)).setHeader("JMSXGroupID",constant("1"))
                        .otherwise().setHeader("JMSXGroupID",constant("2"))
                    .end()
                    .to("amq:queue:Message.Group.Test");

                    /* These two are competing consumers */
                    from("amq:queue:Message.Group.Test").routeId("Route A").log("Received: ${body}");
                    from("amq:queue:Message.Group.Test").routeId("Route B").log("Received: ${body}");
                }
            });

            sendMessages(ctx.createProducerTemplate());
            Thread.sleep(TimeUnit.SECONDS.toMillis(10));
            stopBroker();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static CamelContext createCamelContext() throws Exception {
        CamelContext camelContext = new DefaultCamelContext();

        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("vm://localhost/");
        PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(activeMQConnectionFactory);
        pooledConnectionFactory.setMaxConnections(8);
        pooledConnectionFactory.setMaximumActive(500);

        ActiveMQComponent activeMQComponent = ActiveMQComponent.activeMQComponent();
        activeMQComponent.setUsePooledConnection(true);
        activeMQComponent.setConnectionFactory(pooledConnectionFactory);
        camelContext.addComponent("amq", activeMQComponent);

        return camelContext;
    }

    private static void sendMessages(ProducerTemplate pt) throws Exception {
        for (int i = 0; i < 10; i++) {
            pt.sendBody("direct:begin", Integer.valueOf(i));
        }

        for (int i = 0; i < 10; i++) {
            pt.sendBody("direct:begin", "next group");
        }

        pt.sendBody("direct:begin", Integer.valueOf(1));
        pt.sendBody("direct:begin", "foo");
        pt.sendBody("direct:begin", Integer.valueOf(2));
    }

    private static void startBroker() throws Exception {
        broker = new BrokerService();
        broker.addConnector("vm://localhost");
        broker.start();
    }

    private static void stopBroker() throws Exception {
        broker.stop();
    }
}

The result of running this main method is as follows:

2445 [Camel (camel-1) thread #0 - JmsConsumer[Message.Group.Test]] INFO Route A - Received: 0
2447 [Camel (camel-1) thread #0 - JmsConsumer[Message.Group.Test]] INFO Route A - Received: 1
2460 [Camel (camel-1) thread #0 - JmsConsumer[Message.Group.Test]] INFO Route A - Received: 2
2466 [Camel (camel-1) thread #0 - JmsConsumer[Message.Group.Test]] INFO Route A - Received: 3
2472 [Camel (camel-1) thread #0 - JmsConsumer[Message.Group.Test]] INFO Route A - Received: 4
2479 [Camel (camel-1) thread #0 - JmsConsumer[Message.Group.Test]] INFO Route A - Received: 5
2482 [Camel (camel-1) thread #0 - JmsConsumer[Message.Group.Test]] INFO Route A - Received: 6
2485 [Camel (camel-1) thread #0 - JmsConsumer[Message.Group.Test]] INFO Route A - Received: 7
2488 [Camel (camel-1) thread #0 - JmsConsumer[Message.Group.Test]] INFO Route A - Received: 8
2490 [Camel (camel-1) thread #0 - JmsConsumer[Message.Group.Test]] INFO Route A - Received: 9
2493 [Camel (camel-1) thread #1 - JmsConsumer[Message.Group.Test]] INFO Route B - Received: next group
2496 [Camel (camel-1) thread #1 - JmsConsumer[Message.Group.Test]] INFO Route B - Received: next group
2499 [Camel (camel-1) thread #1 - JmsConsumer[Message.Group.Test]] INFO Route B - Received: next group
2501 [Camel (camel-1) thread #1 - JmsConsumer[Message.Group.Test]] INFO Route B - Received: next group
2504 [Camel (camel-1) thread #1 - JmsConsumer[Message.Group.Test]] INFO Route B - Received: next group
2505 [Camel (camel-1) thread #1 - JmsConsumer[Message.Group.Test]] INFO Route B - Received: next group
2508 [Camel (camel-1) thread #1 - JmsConsumer[Message.Group.Test]] INFO Route B - Received: next group
2510 [Camel (camel-1) thread #1 - JmsConsumer[Message.Group.Test]] INFO Route B - Received: next group
2513 [Camel (camel-1) thread #1 - JmsConsumer[Message.Group.Test]] INFO Route B - Received: next group
2515 [Camel (camel-1) thread #1 - JmsConsumer[Message.Group.Test]] INFO Route B - Received: next group
2517 [Camel (camel-1) thread #0 - JmsConsumer[Message.Group.Test]] INFO Route A - Received: 1
2535 [Camel (camel-1) thread #1 - JmsConsumer[Message.Group.Test]] INFO Route B - Received: foo
2538 [Camel (camel-1) thread #0 - JmsConsumer[Message.Group.Test]] INFO Route A - Received: 2

You’ll notice that all messages with a groupId of 1 are consumed by one route and the messages with a groupId of 2 are consumed by the other consumer. You’ll also see how relatively simple it is to inspect the body of our original message to check it’s type and set the header in the route that begins our orchestration.

If you wish to run this source code, I’ve set up a little Git repository on github for hosting some camel examples. As of the time I write this, only the message group example is available, but others should appear soon.