DZone
Java Zone
Thanks for visiting DZone today,
Edit Profile
  • Manage Email Subscriptions
  • How to Post to DZone
  • Article Submission Guidelines
Sign Out View Profile
  • Post an Article
  • Manage My Drafts
Over 2 million developers have joined DZone.
Log In / Join
  • Refcardz
  • Trend Reports
  • Webinars
  • Zones
  • |
    • Agile
    • AI
    • Big Data
    • Cloud
    • Database
    • DevOps
    • Integration
    • IoT
    • Java
    • Microservices
    • Open Source
    • Performance
    • Security
    • Web Dev
DZone > Java Zone > JMS Message Groups in Apache Camel

JMS Message Groups in Apache Camel

Jason Whaley user avatar by
Jason Whaley
·
Apr. 11, 12 · Java Zone · Interview
Like (0)
Save
Tweet
13.58K Views

Join the DZone community and get the full member experience.

Join For Free

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.

Apache Camel

Published at DZone with permission of Jason Whaley, DZone MVB. See the original article here.

Opinions expressed by DZone contributors are their own.

Popular on DZone

  • Maven Tutorial: Nice and Easy [Video]
  • Use Lambda Function URL To Write a Serverless App Backed by DynamoDB
  • Querying Kafka Topics Using Presto
  • A Simple Guide to Heaps, Stacks, References, and Values in JavaScript

Comments

Java Partner Resources

X

ABOUT US

  • About DZone
  • Send feedback
  • Careers
  • Sitemap

ADVERTISE

  • Advertise with DZone

CONTRIBUTE ON DZONE

  • Article Submission Guidelines
  • MVB Program
  • Become a Contributor
  • Visit the Writers' Zone

LEGAL

  • Terms of Service
  • Privacy Policy

CONTACT US

  • 600 Park Offices Drive
  • Suite 300
  • Durham, NC 27709
  • support@dzone.com
  • +1 (919) 678-0300

Let's be friends:

DZone.com is powered by 

AnswerHub logo