Java Messaging Service (JMS) is all about applications communicating using standard based messaging. With JMS, applications communicates with each other by sending and receiving messages. Spring provides first-class support for application integration using JMS. In this tutorial series, we will explore various ways Spring helps us simplify the JMS integration by abstracting all the complexities and providing simple APIs for use in our applications for synchronous as well as asynchronous communication. Let’s get started.
JMS API Programming Model illustration. Source:Oracle
There are various JMS providers which implements JMS standard and provides additional implementaion-specific functionalities. Apache ActiveMQ , IBM Websphere MQ , JBoss hornetQ to name a few. We will focus on ActiveMQ in this tutorial.
Other interesting posts you may like
- Spring Boot+AngularJS+Spring Data+Hibernate+MySQL CRUD App
- Spring Boot REST API Tutorial
- Spring Boot WAR deployment example
- Spring Boot Introduction + Hello World Example
- Secure Spring REST API using OAuth2
- AngularJS+Spring Security using Basic Authentication
- Secure Spring REST API using Basic Authentication
- Spring 4 MVC+JPA2+Hibernate Many-to-many Example
- Spring 4 Caching Annotations Tutorial
- Spring 4 Cache Tutorial with EhCache
- Spring MVC 4+AngularJS Example
- Spring 4 Email With Attachment Tutorial
- Spring 4 Email Template Library Example
- Spring 4 Email Integration Tutorial
- Spring MVC 4+JMS+ActiveMQ Integration Example
- Spring 4+JMS+ActiveMQ @JmsLister @EnableJms Example
- Spring MVC 4+Hibernate 4 Many-to-many JSP Example
- Spring MVC 4+Hibernate 4+MySQL+Maven integration example using annotations
- Spring MVC4 FileUpload-Download Hibernate+MySQL Example
- TestNG Mockito Integration Example Stubbing Void Methods
- Maven surefire plugin and TestNG Example
- Spring MVC 4 Form Validation and Resource Handling
Following technologies being used:
- Spring 4.3.0.RELEASE
- Spring JMS 4.3.0.RELEASE
- ActiveMQ 5.13.3
- Maven 3
- JDK 1.7
- Tomcat 8.0.21
- Eclipse MARS.1 Release 4.5.1
- logback 1.1.7
Before we jump into code, lets take a quick overview of basic JMS concepts.
Destination
Each JMS message sent by an application is addressed with a destination. Destinations in JMS are like postal mailboxes where the messages are placed until someone comes to pick them up. There are two types of destination in JMS: queues and topics.
-
Queues [point-to-point]: Queue’s are based on point-to-point messaging model where messages are sent to a queue. Each message has exactly one sender and one receiver. Message is guaranteed to be delivered to only one receiver.
JMS Queue destination illustration. Source:Oracle
-
Topics [publish-subscribe]:Topic’s are based on publish-subscribe model where messages are sent to a topic. N subscribers can be subscribed on a topic, and when a message arrives, each will get a copy of that message.
JMS Topic destination illustration. Source:Oracle
ConnectionFactory
In order to create a connection to a messaging provider [message broker], we need connection factories. ConnectionFacotry is basically administrative objects with configurations parameters.
Persistent/Non-persistent Messages [Aka Delivery Mode : for Messages]
Delivery Mode refers to persistence/non-persistence of messages which can be specified on MessageProducer level as well as on individual message level. Default delivery mode is PERSISTENT, means messages will be stored on disk/database until it is consumed by a consumer, and will survive a broker restart. When using non-persistent delivery, if you kill a broker then you will lose all in-transit messages.
Durable / Non-durable subscriptions[for Topics]
Subscription refers to subscription on a topic. With a durable subscription, if the subscriber [which has subscribed for message on a topic] is down for some time, once it comes up, it will receive all the messages sent for it(including the ones sent when it was down). With Non-durable subscription, a subscriber will receive only the messages when it was connected to topic (will loose all the ones sent when it was down). Note that this is not applicable for Queue’s as they can be considered always durable [only one consumer, and it will always receive the message destined for it in queue].
OK, enough with theory, lets get into practice. Firstly download ActiveMQ Message Broker, unzip it, goto bin directory and start it using $ ./activemq start.
You can quickly check the WebConsole [available at http://localhost:8161/admin/ with credentials admin/admin.
Application(s) Overview:
In our post, we have two applications A & B trying to communicate with each other via sending Messages on Queues. Although we could have opted for Topic instead, Queue is perfect for this one producer-one consumer scenario.
A sends a message [a pojo object] to a Queue [order-queue] and listens for the response on another Queue [order-response-queue]. B is listening on order-queue, and upon receiving the message, B will send a reply [a pojo response object] to A on order-response-queue. A short but simple example of inter-application communication using JMS.
Let’s start with coding Application A. Application B is exactly same as A, but listening and sending on opposite queues.
Step 1. Messaging Configuration using Spring & ActiveMQ
package com.websystique.spring.configuration;
import java.util.Arrays;
import javax.jms.ConnectionFactory;
import org.apache.activemq.spring.ActiveMQConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.MessageListenerContainer;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.SimpleMessageConverter;
import com.websystique.spring.messaging.MessageReceiver;
@Configuration
public class MessagingConfiguration {
private static final String DEFAULT_BROKER_URL = "tcp://localhost:61616";
private static final String ORDER_QUEUE = "order-queue";
private static final String ORDER_RESPONSE_QUEUE = "order-response-queue";
@Autowired
MessageReceiver messageReceiver;
@Bean
public ConnectionFactory connectionFactory(){
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
connectionFactory.setBrokerURL(DEFAULT_BROKER_URL);
connectionFactory.setTrustedPackages(Arrays.asList("com.websystique.spring"));
return connectionFactory;
}
/*
* Optionally you can use cached connection factory if performance is a big concern.
*/
@Bean
public ConnectionFactory cachingConnectionFactory(){
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setTargetConnectionFactory(connectionFactory());
connectionFactory.setSessionCacheSize(10);
return connectionFactory;
}
/*
* Message listener container, used for invoking messageReceiver.onMessage on message reception.
*/
@Bean
public MessageListenerContainer getContainer(){
DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setDestinationName(ORDER_RESPONSE_QUEUE);
container.setMessageListener(messageReceiver);
return container;
}
/*
* Used for Sending Messages.
*/
@Bean
public JmsTemplate jmsTemplate(){
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(connectionFactory());
template.setDefaultDestinationName(ORDER_QUEUE);
return template;
}
@Bean
MessageConverter converter(){
return new SimpleMessageConverter();
}
}
ConnectionFactory: ActiveMQ provides a Spring based built-in implementation of JMS connection factory [javax.jms.ConnectionFactory] ActiveMQConnectionFactory, which provides possibilities for further configuration of connection factories. Optionally, if you need more performance, you can configure a CachingConnectionFactory which adds caching.
This connection factory(along with Destination) will be used by both Sender [using JmsTemplate] and receiver [using MessageListenerContainer].
Destination: Destinations needs to be configured for both sending and receiving ends. ActiveMQ comes up with builin implementations for Queue [ActiveMQQueue] and Topic [ActiveMQTopic]which can accept a String [QUEUE or Topic name] as an argument. Although we could use them, we have further simplified our configuration by directly configuring the destination-name on Sending [With JmsTemplate] and Receving [With Listener] side.
JmsTemplate : JmsTemplate provides an abstraction which hides all the complexities of JMS communication. Without JmsTemplate, you will be forced to create connections/sessions/MessageProducers/MessageConsumers and catch all the nasty exception platform may throw. With JmsTemplate ,you get simple API’s to work with , and spring behind-the-scenes take care of all the JMS complexities. It takes care of creating a connection, obtaining a session, and finally sending [as well as synchronous receiving] of message. We will be using JmsTemplate for sending the message. Do note that JmsTemplate also provides possibilities for receiving message but that is synchronous[blocks the listening application], and usually not preferred when asynchronous communication is possible.
MessageListenerContainer : MessageListenerContainer comes handy when we want to implement asynchronous message reception. It can be configured to use a bean [which implements javax.jms.MessageListener] whose onMessage() method will be called on message reception.
Below shown is the bean we have configured for asynchronous Message reception.
package com.websystique.spring.messaging;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.stereotype.Component;
import com.websystique.spring.model.InventoryResponse;
@Component
public class MessageReceiver implements MessageListener{
static final Logger LOG = LoggerFactory.getLogger(MessageReceiver.class);
@Autowired
MessageConverter messageConverter;
@Override
public void onMessage(Message message) {
try {
LOG.info("+++++++++++++++++++++++++++++++++++++++++++++++++++++");
InventoryResponse response = (InventoryResponse) messageConverter.fromMessage(message);
LOG.info("Application : order response received : {}",response);
LOG.info("+++++++++++++++++++++++++++++++++++++++++++++++++++++");
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
Thanks to MessageListenerContainer, onMessage method of MessageReceiver bean will be called whenever there is a message on the queue [The Listener was listening to].
MessageConverter : Spring further helps us by providing converters which convert Java objects to JMS messages and viceversa.
We have seen the receiving side, let’s see how to send the Message now. JmsTemplate is all you need to send the messages. It comes with several methods[send*,convertAndSend*] to choose from while sending the messages.
package com.websystique.spring.messaging;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;
import com.websystique.spring.model.Product;
@Component
public class MessageSender {
@Autowired
JmsTemplate jmsTemplate;
public void sendMessage(final Product product) {
jmsTemplate.send(new MessageCreator(){
@Override
public Message createMessage(Session session) throws JMSException{
ObjectMessage objectMessage = session.createObjectMessage(product);
return objectMessage;
}
});
}
}
That’s all for Application A setup. Below shown is the Directory Structure for Project A.
Now just flip the QUEUE’s and Application B is ready. Before we start both the applications and see message transfer in action, shown below is the directory structure and configuration code for Application B [Mirror image].
package com.websystique.spring.configuration;
import java.util.Arrays;
import javax.jms.ConnectionFactory;
import org.apache.activemq.spring.ActiveMQConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.MessageListenerContainer;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.SimpleMessageConverter;
import com.websystique.spring.messaging.MessageReceiver;
@Configuration
public class MessagingConfiguration {
private static final String DEFAULT_BROKER_URL = "tcp://localhost:61616";
private static final String ORDER_QUEUE = "order-queue";
private static final String ORDER_RESPONSE_QUEUE = "order-response-queue";
@Autowired
MessageReceiver messageReceiver;
@Bean
public ConnectionFactory connectionFactory(){
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
connectionFactory.setBrokerURL(DEFAULT_BROKER_URL);
connectionFactory.setTrustedPackages(Arrays.asList("com.websystique.spring"));
return connectionFactory;
}
/*
* Unused.
*/
@Bean
public ConnectionFactory cachingConnectionFactory(){
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setTargetConnectionFactory(connectionFactory());
connectionFactory.setSessionCacheSize(10);
return connectionFactory;
}
/*
* Message listener container, used for invoking messageReceiver.onMessage on message reception.
*/
@Bean
public MessageListenerContainer getContainer(){
DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setDestinationName(ORDER_QUEUE);
container.setMessageListener(messageReceiver);
return container;
}
/*
* Used for Sending Messages.
*/
@Bean
public JmsTemplate jmsTemplate(){
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(connectionFactory());
template.setDefaultDestinationName(ORDER_RESPONSE_QUEUE);
return template;
}
@Bean
MessageConverter converter(){
return new SimpleMessageConverter();
}
}
package com.websystique.spring.messaging;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.stereotype.Component;
import com.websystique.spring.model.Product;
import com.websystique.spring.service.OrderService;
@Component
public class MessageReceiver implements MessageListener{
static final Logger LOG = LoggerFactory.getLogger(MessageReceiver.class);
@Autowired
MessageConverter messageConverter;
@Autowired
OrderService orderService;
@Override
public void onMessage(Message message) {
try {
LOG.info("-----------------------------------------------------");
Product product = (Product) messageConverter.fromMessage(message);
LOG.info("Application : order received : {}",product);
orderService.processOrder(product);
LOG.info("-----------------------------------------------------");
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
package com.websystique.spring.messaging;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;
import com.websystique.spring.model.InventoryResponse;
@Component
public class MessageSender {
@Autowired
JmsTemplate jmsTemplate;
public void sendMessage(final InventoryResponse inventoryResponse) {
jmsTemplate.send(new MessageCreator(){
@Override
public Message createMessage(Session session) throws JMSException{
ObjectMessage objectMessage = session.createObjectMessage(inventoryResponse);
return objectMessage;
}
});
}
}
That is all.
package com.websystique.spring.configuration;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
import com.websystique.spring.model.Product;
import com.websystique.spring.service.ProductService;
import com.websystique.spring.util.BasicUtil;
public class ProducerApplication {
static final Logger LOG = LoggerFactory.getLogger(ProducerApplication.class);
private static AtomicInteger id = new AtomicInteger();
public static void main(String[] args){
AbstractApplicationContext context = new AnnotationConfigApplicationContext(
AppConfig.class);
ProductService productService = (ProductService) context.getBean("productService");
Product product = getProduct();
LOG.info("Application : sending order {}", product);
productService.sendProduct(product);
try {
Thread.sleep(60000);
} catch (InterruptedException e) {
e.printStackTrace();
}
((AbstractApplicationContext)context).close();
}
private static Product getProduct(){
Product p = new Product();
p.setName("Product "+id.incrementAndGet());
p.setProductId(BasicUtil.getUniqueId());
p.setQuantity(2);
return p;
}
}
package com.websystique.spring.configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
public class ConsumerApplication {
static final Logger LOG = LoggerFactory.getLogger(ConsumerApplication.class);
public static void main(String[] args) {
AbstractApplicationContext context = new AnnotationConfigApplicationContext(AppConfig.class);
try {
Thread.sleep(60000);
} catch (InterruptedException e) {
e.printStackTrace();
}
((AbstractApplicationContext) context).close();
}
}
Start Application A [ProducerEx1]. Goto web-console, and click on queue tab. You should see two queues being created with a message on one queue [A has just sent a message] while other empty.
Now start Application B [ConsumerEx1], again check queues on web-console, you will see message being de-queued on first queue [B has just processed it] and new message being enqueued [B has sent the response to A] and eventually dequeued [A got the response] on second queue.
Logs of Producer Application:
Jun 26, 2016 1:50:32 AM org.springframework.context.annotation.AnnotationConfigApplicationContext prepareRefresh
INFO: Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@51f38d7d: startup date [Sun Jun 26 01:50:32 CEST 2016]; root of context hierarchy
Jun 26, 2016 1:50:33 AM org.springframework.context.support.DefaultLifecycleProcessor start
INFO: Starting beans in phase 2147483647
01:50:33.470 [main] DEBUG org.apache.activemq.transport.WireFormatNegotiator - Sending: WireFormatInfo { version=11, properties={MaxFrameSize=9223372036854775807, CacheSize=1024, CacheEnabled=true, Host=localhost, SizePrefixDisabled=false, MaxInactivityDurationInitalDelay=10000, TcpNoDelayEnabled=true, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]}
01:50:33.472 [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62424] DEBUG org.apache.activemq.transport.WireFormatNegotiator - Received WireFormat: WireFormatInfo { version=11, properties={CacheSize=1024, MaxFrameSize=104857600, CacheEnabled=true, SizePrefixDisabled=false, TcpNoDelayEnabled=true, MaxInactivityDurationInitalDelay=10000, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]}
01:50:33.476 [main] DEBUG org.apache.activemq.transport.InactivityMonitor - Using min of local: WireFormatInfo { version=11, properties={MaxFrameSize=9223372036854775807, CacheSize=1024, CacheEnabled=true, Host=localhost, SizePrefixDisabled=false, MaxInactivityDurationInitalDelay=10000, TcpNoDelayEnabled=true, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]} and remote: WireFormatInfo { version=11, properties={CacheSize=1024, MaxFrameSize=104857600, CacheEnabled=true, SizePrefixDisabled=false, TcpNoDelayEnabled=true, MaxInactivityDurationInitalDelay=10000, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]}
01:50:33.477 [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62424] DEBUG org.apache.activemq.transport.WireFormatNegotiator - tcp://localhost/127.0.0.1:61616@62424 before negotiation: OpenWireFormat{version=11, cacheEnabled=false, stackTraceEnabled=false, tightEncodingEnabled=false, sizePrefixDisabled=false, maxFrameSize=9223372036854775807}
01:50:33.478 [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62424] DEBUG org.apache.activemq.transport.WireFormatNegotiator - tcp://localhost/127.0.0.1:61616@62424 after negotiation: OpenWireFormat{version=11, cacheEnabled=true, stackTraceEnabled=true, tightEncodingEnabled=true, sizePrefixDisabled=false, maxFrameSize=104857600}
01:50:33.580 [getContainer-1] DEBUG org.apache.activemq.thread.TaskRunnerFactory - Initialized TaskRunnerFactory[ActiveMQ Session Task] using ExecutorService: java.util.concurrent.ThreadPoolExecutor@33a2215d[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
01:50:33.654 [main] INFO com.websystique.spring.configuration.ProducerApplication - Application : sending order Product [productId=d4223424-9aac-467a-93c8-aa357c0d524f, name=Product 1, quantity=2]
01:50:33.654 [main] INFO com.websystique.spring.service.ProductServiceImpl - +++++++++++++++++++++++++++++++++++++++++++++++++++++
01:50:33.655 [main] INFO com.websystique.spring.service.ProductServiceImpl - Application : sending order request Product [productId=d4223424-9aac-467a-93c8-aa357c0d524f, name=Product 1, quantity=2]
01:50:33.662 [main] DEBUG org.apache.activemq.transport.WireFormatNegotiator - Sending: WireFormatInfo { version=11, properties={MaxFrameSize=9223372036854775807, CacheSize=1024, CacheEnabled=true, Host=localhost, SizePrefixDisabled=false, MaxInactivityDurationInitalDelay=10000, TcpNoDelayEnabled=true, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]}
01:50:33.666 [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62425] DEBUG org.apache.activemq.transport.InactivityMonitor - Using min of local: WireFormatInfo { version=11, properties={MaxFrameSize=9223372036854775807, CacheSize=1024, CacheEnabled=true, Host=localhost, SizePrefixDisabled=false, MaxInactivityDurationInitalDelay=10000, TcpNoDelayEnabled=true, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]} and remote: WireFormatInfo { version=11, properties={CacheSize=1024, MaxFrameSize=104857600, CacheEnabled=true, SizePrefixDisabled=false, TcpNoDelayEnabled=true, MaxInactivityDurationInitalDelay=10000, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]}
01:50:33.666 [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62425] DEBUG org.apache.activemq.transport.WireFormatNegotiator - Received WireFormat: WireFormatInfo { version=11, properties={CacheSize=1024, MaxFrameSize=104857600, CacheEnabled=true, SizePrefixDisabled=false, TcpNoDelayEnabled=true, MaxInactivityDurationInitalDelay=10000, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]}
01:50:33.666 [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62425] DEBUG org.apache.activemq.transport.WireFormatNegotiator - tcp://localhost/127.0.0.1:61616@62425 before negotiation: OpenWireFormat{version=11, cacheEnabled=false, stackTraceEnabled=false, tightEncodingEnabled=false, sizePrefixDisabled=false, maxFrameSize=9223372036854775807}
01:50:33.666 [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62425] DEBUG org.apache.activemq.transport.WireFormatNegotiator - tcp://localhost/127.0.0.1:61616@62425 after negotiation: OpenWireFormat{version=11, cacheEnabled=true, stackTraceEnabled=true, tightEncodingEnabled=true, sizePrefixDisabled=false, maxFrameSize=104857600}
01:50:33.802 [main] DEBUG org.apache.activemq.util.ThreadPoolUtils - Shutdown of ExecutorService: java.util.concurrent.ThreadPoolExecutor@60b419dd[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0] is shutdown: true and terminated: true took: 0.000 seconds.
01:50:33.803 [main] DEBUG org.apache.activemq.transport.tcp.TcpTransport - Stopping transport tcp://localhost/127.0.0.1:61616@62425
01:50:33.804 [main] DEBUG org.apache.activemq.thread.TaskRunnerFactory - Initialized TaskRunnerFactory[ActiveMQ Task] using ExecutorService: java.util.concurrent.ThreadPoolExecutor@43f42525[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
01:50:33.805 [ActiveMQ Task-1] DEBUG org.apache.activemq.transport.tcp.TcpTransport - Closed socket Socket[addr=localhost/127.0.0.1,port=61616,localport=62425]
01:50:33.806 [main] DEBUG org.apache.activemq.util.ThreadPoolUtils - Forcing shutdown of ExecutorService: java.util.concurrent.ThreadPoolExecutor@43f42525[Running, pool size = 1, active threads = 0, queued tasks = 0, completed tasks = 1]
01:50:33.806 [main] INFO com.websystique.spring.service.ProductServiceImpl - +++++++++++++++++++++++++++++++++++++++++++++++++++++
01:50:53.478 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
01:50:53.480 [ActiveMQ InactivityMonitor Worker] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
01:51:03.478 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - WriteChecker: 10000ms elapsed since last write check.
01:51:03.478 [ActiveMQ InactivityMonitor Worker] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
01:51:13.478 [ActiveMQ InactivityMonitor ReadCheckTimer] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - 30001ms elapsed since last read check.
01:51:13.479 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
01:51:13.479 [ActiveMQ InactivityMonitor Worker] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
01:51:17.460 [getContainer-1] INFO com.websystique.spring.messaging.MessageReceiver - +++++++++++++++++++++++++++++++++++++++++++++++++++++
01:51:17.464 [getContainer-1] INFO com.websystique.spring.messaging.MessageReceiver - Application : order response received : InventoryResponse [productId=d4223424-9aac-467a-93c8-aa357c0d524f, returnCode=200, comment=Order Processed successfully]
01:51:17.464 [getContainer-1] INFO com.websystique.spring.messaging.MessageReceiver - +++++++++++++++++++++++++++++++++++++++++++++++++++++
01:51:23.479 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - WriteChecker: 10000ms elapsed since last write check.
01:51:33.480 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
01:51:33.480 [ActiveMQ InactivityMonitor Worker] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
Jun 26, 2016 1:51:33 AM org.springframework.context.annotation.AnnotationConfigApplicationContext doClose
INFO: Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@51f38d7d: startup date [Sun Jun 26 01:50:32 CEST 2016]; root of context hierarchy
Jun 26, 2016 1:51:33 AM org.springframework.context.support.DefaultLifecycleProcessor stop
INFO: Stopping beans in phase 2147483647
01:51:34.465 [getContainer-1] DEBUG org.apache.activemq.ActiveMQMessageConsumer - remove: ID:dragon-62423-1466898633347-1:1:1:1, lastDeliveredSequenceId: 47
01:51:34.512 [main] DEBUG org.apache.activemq.util.ThreadPoolUtils - Shutdown of ExecutorService: java.util.concurrent.ThreadPoolExecutor@33a2215d[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 2] is shutdown: true and terminated: false took: 0.001 seconds.
01:51:34.512 [main] DEBUG org.apache.activemq.util.ThreadPoolUtils - Shutdown of ExecutorService: java.util.concurrent.ThreadPoolExecutor@65780f1[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0] is shutdown: true and terminated: true took: 0.000 seconds.
01:51:34.512 [main] DEBUG org.apache.activemq.transport.tcp.TcpTransport - Stopping transport tcp://localhost/127.0.0.1:61616@62424
01:51:34.513 [main] DEBUG org.apache.activemq.thread.TaskRunnerFactory - Initialized TaskRunnerFactory[ActiveMQ Task] using ExecutorService: java.util.concurrent.ThreadPoolExecutor@5ad25177[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
01:51:34.514 [ActiveMQ Task-1] DEBUG org.apache.activemq.transport.tcp.TcpTransport - Closed socket Socket[addr=localhost/127.0.0.1,port=61616,localport=62424]
01:51:34.514 [main] DEBUG org.apache.activemq.util.ThreadPoolUtils - Forcing shutdown of ExecutorService: java.util.concurrent.ThreadPoolExecutor@5ad25177[Running, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]
Logs of Consumer Application:
Jun 26, 2016 1:51:16 AM org.springframework.context.annotation.AnnotationConfigApplicationContext prepareRefresh
INFO: Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@117280a: startup date [Sun Jun 26 01:51:16 CEST 2016]; root of context hierarchy
Jun 26, 2016 1:51:17 AM org.springframework.context.support.DefaultLifecycleProcessor start
INFO: Starting beans in phase 2147483647
01:51:17.342 [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62428] DEBUG org.apache.activemq.transport.WireFormatNegotiator - Received WireFormat: WireFormatInfo { version=11, properties={CacheSize=1024, MaxFrameSize=104857600, CacheEnabled=true, SizePrefixDisabled=false, TcpNoDelayEnabled=true, MaxInactivityDurationInitalDelay=10000, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]}
01:51:17.340 [main] DEBUG org.apache.activemq.transport.WireFormatNegotiator - Sending: WireFormatInfo { version=11, properties={MaxFrameSize=9223372036854775807, CacheSize=1024, CacheEnabled=true, Host=localhost, SizePrefixDisabled=false, MaxInactivityDurationInitalDelay=10000, TcpNoDelayEnabled=true, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]}
01:51:17.347 [main] DEBUG org.apache.activemq.transport.InactivityMonitor - Using min of local: WireFormatInfo { version=11, properties={MaxFrameSize=9223372036854775807, CacheSize=1024, CacheEnabled=true, Host=localhost, SizePrefixDisabled=false, MaxInactivityDurationInitalDelay=10000, TcpNoDelayEnabled=true, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]} and remote: WireFormatInfo { version=11, properties={CacheSize=1024, MaxFrameSize=104857600, CacheEnabled=true, SizePrefixDisabled=false, TcpNoDelayEnabled=true, MaxInactivityDurationInitalDelay=10000, MaxInactivityDuration=30000, TightEncodingEnabled=true, StackTraceEnabled=true}, magic=[A,c,t,i,v,e,M,Q]}
01:51:17.349 [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62428] DEBUG org.apache.activemq.transport.WireFormatNegotiator - tcp://localhost/127.0.0.1:61616@62428 before negotiation: OpenWireFormat{version=11, cacheEnabled=false, stackTraceEnabled=false, tightEncodingEnabled=false, sizePrefixDisabled=false, maxFrameSize=9223372036854775807}
01:51:17.350 [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62428] DEBUG org.apache.activemq.transport.WireFormatNegotiator - tcp://localhost/127.0.0.1:61616@62428 after negotiation: OpenWireFormat{version=11, cacheEnabled=true, stackTraceEnabled=true, tightEncodingEnabled=true, sizePrefixDisabled=false, maxFrameSize=104857600}
01:51:17.413 [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62428] DEBUG org.apache.activemq.thread.TaskRunnerFactory - Initialized TaskRunnerFactory[ActiveMQ Session Task] using ExecutorService: java.util.concurrent.ThreadPoolExecutor@74131e8[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
01:51:17.422 [getContainer-1] INFO com.websystique.spring.messaging.MessageReceiver - -----------------------------------------------------
01:51:17.433 [getContainer-1] INFO com.websystique.spring.messaging.MessageReceiver - Application : order received : Product [productId=d4223424-9aac-467a-93c8-aa357c0d524f, name=Product 1, quantity=2]
01:51:17.434 [getContainer-1] INFO com.websystique.spring.service.OrderServiceImpl - Inventory : sending order confirmation InventoryResponse [productId=d4223424-9aac-467a-93c8-aa357c0d524f, returnCode=200, comment=Order Processed successfully]
01:51:17.530 [getContainer-1] INFO com.websystique.spring.messaging.MessageReceiver - -----------------------------------------------------
01:51:37.349 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
01:51:37.351 [ActiveMQ InactivityMonitor Worker] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
01:51:47.350 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
01:51:47.350 [ActiveMQ InactivityMonitor Worker] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
01:51:57.349 [ActiveMQ InactivityMonitor ReadCheckTimer] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - 30001ms elapsed since last read check.
01:51:57.350 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - WriteChecker: 10000ms elapsed since last write check.
01:51:57.350 [ActiveMQ InactivityMonitor Worker] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
01:52:07.351 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - WriteChecker: 10001ms elapsed since last write check.
01:52:07.351 [ActiveMQ InactivityMonitor Worker] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
01:52:17.351 [ActiveMQ InactivityMonitor WriteCheckTimer] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - WriteChecker: 10000ms elapsed since last write check.
01:52:17.351 [ActiveMQ InactivityMonitor Worker] DEBUG org.apache.activemq.transport.AbstractInactivityMonitor - Running WriteCheck[tcp://127.0.0.1:61616]
Jun 26, 2016 1:52:17 AM org.springframework.context.annotation.AnnotationConfigApplicationContext doClose
INFO: Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@117280a: startup date [Sun Jun 26 01:51:16 CEST 2016]; root of context hierarchy
Jun 26, 2016 1:52:17 AM org.springframework.context.support.DefaultLifecycleProcessor stop
INFO: Stopping beans in phase 2147483647
01:52:17.539 [main] DEBUG org.apache.activemq.ActiveMQMessageConsumer - remove: ID:dragon-62427-1466898677206-1:1:1:1, lastDeliveredSequenceId: 41
01:52:17.552 [main] DEBUG org.apache.activemq.util.ThreadPoolUtils - Shutdown of ExecutorService: java.util.concurrent.ThreadPoolExecutor@74131e8[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 2] is shutdown: true and terminated: true took: 0.000 seconds.
01:52:17.553 [main] DEBUG org.apache.activemq.util.ThreadPoolUtils - Shutdown of ExecutorService: java.util.concurrent.ThreadPoolExecutor@718316f2[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0] is shutdown: true and terminated: true took: 0.000 seconds.
01:52:17.554 [main] DEBUG org.apache.activemq.transport.tcp.TcpTransport - Stopping transport tcp://localhost/127.0.0.1:61616@62428
01:52:17.556 [main] DEBUG org.apache.activemq.thread.TaskRunnerFactory - Initialized TaskRunnerFactory[ActiveMQ Task] using ExecutorService: java.util.concurrent.ThreadPoolExecutor@26151aae[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
01:52:17.557 [ActiveMQ Task-1] DEBUG org.apache.activemq.transport.tcp.TcpTransport - Closed socket Socket[addr=localhost/127.0.0.1,port=61616,localport=62428]
01:52:17.557 [main] DEBUG org.apache.activemq.util.ThreadPoolUtils - Forcing shutdown of ExecutorService: java.util.concurrent.ThreadPoolExecutor@26151aae[Running, pool size = 1, active threads = 0, queued tasks = 0, completed tasks = 1]
For the sake of completeness, shown below are the side-configuration/classes needed to run the examples.
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.websystique.spring</groupId>
<artifactId>SpringJMSActiveMQProducerEx1</artifactId>
<version>1.0.0</version>
<packaging>jar</packaging>
<name>SpringJMSActiveMQProducerEx1</name>
<properties>
<springframework.version>4.3.0.RELEASE</springframework.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<version>${springframework.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${springframework.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${springframework.version}</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-spring</artifactId>
<version>5.13.3</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.1.7</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.2</version>
<configuration>
<source>1.7</source>
<target>1.7</target>
</configuration>
</plugin>
</plugins>
</build>
</project>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.websystique.spring</groupId>
<artifactId>SpringJMSActiveMQConsumerEx1</artifactId>
<version>1.0.0</version>
<packaging>jar</packaging>
<name>SpringJMSActiveMQConsumerEx1</name>
<properties>
<springframework.version>4.3.0.RELEASE</springframework.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<version>${springframework.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>${springframework.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${springframework.version}</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-spring</artifactId>
<version>5.13.3</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.1.7</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.2</version>
<configuration>
<source>1.7</source>
<target>1.7</target>
</configuration>
</plugin>
</plugins>
</build>
</project>
package com.websystique.spring.configuration;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
@Configuration
@ComponentScan(basePackages = "com.websystique.spring")
@Import({MessagingConfiguration.class})
public class AppConfig {
//Put Other Application configuration here.
}
package com.websystique.spring.model;
import java.io.Serializable;
public class Product implements Serializable {
private String productId;
private String name;
private int quantity;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getQuantity() {
return quantity;
}
public void setQuantity(int quantity) {
this.quantity = quantity;
}
public String getProductId() {
return productId;
}
public void setProductId(String productId) {
this.productId = productId;
}
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result + ((productId == null) ? 0 : productId.hashCode());
return result;
}
@Override
public boolean equals(Object obj) {
if (this == obj)
return true;
if (obj == null)
return false;
if (getClass() != obj.getClass())
return false;
Product other = (Product) obj;
if (productId == null) {
if (other.productId != null)
return false;
} else if (!productId.equals(other.productId))
return false;
return true;
}
@Override
public String toString() {
return "Product [productId=" + productId + ", name=" + name + ", quantity=" + quantity + "]";
}
}
package com.websystique.spring.model;
import java.io.Serializable;
public class InventoryResponse implements Serializable{
private String productId;
private int returnCode;
private String comment;
public String getProductId() {
return productId;
}
public void setProductId(String productId) {
this.productId = productId;
}
public int getReturnCode() {
return returnCode;
}
public void setReturnCode(int returnCode) {
this.returnCode = returnCode;
}
public String getComment() {
return comment;
}
public void setComment(String comment) {
this.comment = comment;
}
@Override
public String toString() {
return "InventoryResponse [productId=" + productId + ", returnCode=" + returnCode + ", comment=" + comment
+ "]";
}
}
package com.websystique.spring.service;
import com.websystique.spring.model.Product;
public interface OrderService {
public void processOrder(Product product);
}
package com.websystique.spring.service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.websystique.spring.messaging.MessageSender;
import com.websystique.spring.model.InventoryResponse;
import com.websystique.spring.model.Product;
@Service("orderService")
public class OrderServiceImpl implements OrderService{
static final Logger LOG = LoggerFactory.getLogger(OrderServiceImpl.class);
@Autowired
MessageSender messageSender;
@Override
public void processOrder(Product product) {
InventoryResponse response = prepareResponse(product);
LOG.info("Inventory : sending order confirmation {}", response);
messageSender.sendMessage(response);
}
private InventoryResponse prepareResponse(Product product){
InventoryResponse response = new InventoryResponse();
response.setProductId(product.getProductId());
response.setReturnCode(200);
response.setComment("Order Processed successfully");
return response;
}
}
package com.websystique.spring.service;
import com.websystique.spring.model.Product;
public interface ProductService {
public void sendProduct(Product product);
}
package com.websystique.spring.service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.websystique.spring.messaging.MessageSender;
import com.websystique.spring.model.Product;
@Service("productService")
public class ProductServiceImpl implements ProductService{
static final Logger LOG = LoggerFactory.getLogger(ProductServiceImpl.class);
@Autowired
MessageSender messageSender;
@Override
public void sendProduct(Product product) {
LOG.info("+++++++++++++++++++++++++++++++++++++++++++++++++++++");
LOG.info("Application : sending order request {}", product);
messageSender.sendMessage(product);
LOG.info("+++++++++++++++++++++++++++++++++++++++++++++++++++++");
}
}
package com.websystique.spring.util;
import java.util.UUID;
public class BasicUtil {
public static String getUniqueId(){
return UUID.randomUUID().toString();
}
}
That’s it. As we saw, SPRING JMS support is easy to use. Feel free to Comment, and suggest improvements.
Download Source Code
Producer Application :
Consumer Application :
References
If you like tutorials on this site, why not take a step further and connect me on Facebook , Google Plus & Twitter as well? I would love to hear your thoughts on these articles, it will help improve further our learning process.