欢迎光临
我们一直在努力

Spring JMS ActiveMQ Queue Example

Java Messaging Service (JMS) is all about applications communication 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.

Spring  JMS ActiveMQ Queue Example
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.


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.
    Spring  JMS ActiveMQ Queue Example

    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.
    Spring  JMS ActiveMQ Queue Example
    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 downloadActiveMQ Message Broker, unzip it, goto bin directory and start it using$ ./activemq
start
.

You can quickly check the WebConsole [available athttp://localhost:8161/admin/ with credentialsadmin/admin.
Spring  JMS ActiveMQ Queue Example

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 onorder-queue, and upon receiving
the message, B will send a reply [a pojo response object] to A onorder-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

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"  
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context"    
    xmlns:mongo="http://www.springframework.org/schema/data/mongo"  
    xmlns:aop="http://www.springframework.org/schema/aop"
    xsi:schemaLocation="http://www.springframework.org/schema/beans   
            http://www.springframework.org/schema/beans/spring-beans-3.0.xsd  
            http://www.springframework.org/schema/data/mongo  
            http://www.springframework.org/schema/data/mongo/spring-mongo.xsd
            http://www.springframework.org/schema/context  
     	http://www.springframework.org/schema/context/spring-context-3.0.xsd
     	http://www.springframework.org/schema/aop
        http://www.springframework.org/schema/aop/spring-aop-3.0.xsd">  



	<context:component-scan base-package="com.npf"/>
	
	
	<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
		<property name="brokerURL" value="tcp://localhost:61616"/>
	</bean>
	
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory" ref="connectionFactory"/>
		<property name="defaultDestinationName" value="order-queue"/>
	</bean>
	
	<bean id="messageListenerContainer " class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="connectionFactory"/>
		<property name="destinationName" value="order-response-queue"/>
		<property name="messageListener" ref="messageReceiver"/>
	</bean>
	
	<bean id="messageConverter" class="org.springframework.jms.support.converter.SimpleMessageConverter"/>


</beans>

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 aCachingConnectionFactory 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 implementsjavax.jms.MessageListener] whoseonMessage() method will be called on message reception.

Below shown is the bean we have configured for asynchronous Message reception.

package com.npf.receiver;

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.fasterxml.jackson.core.type.TypeReference;
import com.npf.model.InventoryResponse;
import com.npf.util.JsonUtil;


@Component("messageReceiver")
public class MessageReceiver implements MessageListener{
 
    static final Logger LOG = LoggerFactory.getLogger(MessageReceiver.class);
     
    @Autowired
    private MessageConverter messageConverter;
     
    @SuppressWarnings("unused")
	@Override
    public void onMessage(Message message) {
        try {
        	String json = (String) messageConverter.fromMessage(message);
        	LOG.info("PRODUCER response received : {}",json);
        	InventoryResponse response = JsonUtil.convertJsonToBean(json, new TypeReference<InventoryResponse>(){});
        } catch (JMSException e) {
           
        }
         
    }
}


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.npf.service.impl;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;

import com.npf.model.Product;
import com.npf.service.ProductService;
import com.npf.util.JsonUtil;

@Service("productService")
public class ProductServiceImpl implements ProductService{

	static final Logger LOG = LoggerFactory.getLogger(ProductServiceImpl.class);
	
	@Autowired
	private JmsTemplate jmsTemplate;
	
	@Override
	public void sendProduct(Product product) {
		final String json = JsonUtil.converBeanToJson(product);
		LOG.info("PRODUCER sending request {}", json);
		 jmsTemplate.send(new MessageCreator(){
             @Override
             public Message createMessage(Session session) throws JMSException{
                 ObjectMessage objectMessage = session.createObjectMessage(json);
                 return objectMessage;
             }
         });
	}

	
}


That’s all for Application A setup. Below shown is the Directory Structure for Project A.

Spring  JMS ActiveMQ Queue Example


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].

Spring  JMS ActiveMQ Queue Example

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"  
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context"    
    xmlns:mongo="http://www.springframework.org/schema/data/mongo"  
    xmlns:aop="http://www.springframework.org/schema/aop"
    xsi:schemaLocation="http://www.springframework.org/schema/beans   
            http://www.springframework.org/schema/beans/spring-beans-3.0.xsd  
            http://www.springframework.org/schema/data/mongo  
            http://www.springframework.org/schema/data/mongo/spring-mongo.xsd
            http://www.springframework.org/schema/context  
     	http://www.springframework.org/schema/context/spring-context-3.0.xsd
     	http://www.springframework.org/schema/aop
        http://www.springframework.org/schema/aop/spring-aop-3.0.xsd">  



	<context:component-scan base-package="com.npf"/>
	
	
	<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
		<property name="brokerURL" value="tcp://localhost:61616"/>
	</bean>
	
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory" ref="connectionFactory"/>
		<property name="defaultDestinationName" value="order-response-queue"/>
	</bean>
	
	<bean id="messageListenerContainer " class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="connectionFactory"/>
		<property name="destinationName" value="order-queue"/>
		<property name="messageListener" ref="messageReceiver"/>
	</bean>
	
	<bean id="messageConverter" class="org.springframework.jms.support.converter.SimpleMessageConverter"/>


</beans>


package com.npf.receiver;

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.fasterxml.jackson.core.type.TypeReference;
import com.npf.model.Product;
import com.npf.util.JsonUtil;


@Component
public class MessageReceiver implements MessageListener{
 
    static final Logger LOG = LoggerFactory.getLogger(MessageReceiver.class);
     
    @Autowired
    private MessageConverter messageConverter;
 
     
    @SuppressWarnings("unused")
	@Override
    public void onMessage(Message message) {
        try {
            String json = (String) messageConverter.fromMessage(message);
            LOG.info("CONSUMER : received : {}",json);  
            Product product = JsonUtil.convertJsonToBean(json, new TypeReference<Product>(){});
        } catch (JMSException e) {
            
        }
         
    }
}


package com.npf.send;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.npf.model.InventoryResponse;
import com.npf.util.JsonUtil;
 
@Component
public class MessageSender {
 
	static final Logger LOG = LoggerFactory.getLogger(MessageSender.class);
	
    @Autowired
    private JmsTemplate jmsTemplate;
 
    public void sendMessage(final InventoryResponse inventoryResponse) {
    	
    	final String json = JsonUtil.converBeanToJson(inventoryResponse);
    	
        jmsTemplate.send(new MessageCreator(){
                @Override
                public Message createMessage(Session session) throws JMSException{
                    ObjectMessage objectMessage = session.createObjectMessage(json);
                    LOG.info("CONSUMER : send : {}",json);    
                    return objectMessage;
                }
            });
    }
 
}


That is all.

package com.npf.test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.npf.model.Product;
import com.npf.service.ProductService;
 
public class ProducerApplication {
 
    static final Logger LOG = LoggerFactory.getLogger(ProducerApplication.class);

     
    @SuppressWarnings("resource")
	public static void main(String[] args){
        ApplicationContext context = new ClassPathXmlApplicationContext("spring/applicationContext.xml");
        ProductService productService = (ProductService) context.getBean("productService");
        Product product = new Product();
        product.setName("pingguo");
        product.setProductId("1");
        product.setQuantity(1);
        productService.sendProduct(product);
        try {
            Thread.sleep(60000);
        } catch (InterruptedException e) {
            
        }
    }
}


package com.npf.test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.npf.model.InventoryResponse;
import com.npf.send.MessageSender;
 
public class ConsumerApplication {
 
    static final Logger LOG = LoggerFactory.getLogger(ConsumerApplication.class);
 
    @SuppressWarnings({ "resource" })
	public static void main(String[] args) {
    	 ApplicationContext context = new ClassPathXmlApplicationContext("spring/applicationContext.xml");
        try {
            Thread.sleep(60000);
        } catch (InterruptedException e) {
           
        }
        
        MessageSender messageSender = context.getBean(MessageSender.class);
        InventoryResponse inventoryResponse = new InventoryResponse();
        inventoryResponse.setProductId("1");
        inventoryResponse.setComment("hhh");
        inventoryResponse.setReturnCode(200);
        messageSender.sendMessage(inventoryResponse);
    }
 
}


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.

Spring  JMS ActiveMQ Queue Example

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.

Spring  JMS ActiveMQ Queue Example
Logs of Producer Application:

Spring  JMS ActiveMQ Queue Example


Logs of Consumer Application:

Spring  JMS ActiveMQ Queue Example

That’s it. As we saw, SPRING JMS support is easy to use. Feel free to Comment, and suggest improvements.

Producer Application :https://github.com/SpringOrganization/springJMSproducer

Consumer
Application :https://github.com/SpringOrganization/springJMSConsumer

ref link : http://websystique.com/spring/spring-4-jms-activemq-example-with-annotations/

该文章由WP-AutoPost插件自动采集发布

原文地址:http://blog.csdn.net/pfnie/article/details/51830577

未经允许不得转载:SRE空间 » Spring JMS ActiveMQ Queue Example

分享到:更多 ()

评论 抢沙发

评论前必须登录!

 

oracle

联系我们联系我们