JMS之Spring +activeMQ实现消息队列

henu_zhangyang

浏览: 338559 次
性别:
来自: 开封

最近访客更多访客>>

apex53

ptkr

sampou

woailihui

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

jms activeMQ Spring

JMS

JMS的全称是Java Message Service，即Java消息服务。它主要用于在生产者和消费者之间进行消息传递，生产者负责产生消息，而消费者负责接收消息。把它应用到实际的业务需求中的话我们可以在特定的时候利用生产者生成一消息，并进行发送，对应的消费者在接收到对应的消息后去完成对应的业务逻辑。对于消息的传递有两种类型，一种是点对点的，即一个生产者和一个消费者一一对应；另一种是发布/订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收

本质：通过JMS消息服务器，使生产者和消费者之间解耦。这种技术类似于JDBC，同样是一种规范，各个厂商之间对JMS规范进行实现，在使用时，导入相应厂商的实现类，同类产品有activeMQ、JbossMQ、kafka等

本文通过Spring集成activeMQ实现一个producer程序、和consumer程序，两者通过MQ进行通讯，具体如下：

1.producer工程和consumer工程目录图

2.maven依赖

	<dependencies>
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>4.10</version>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-context</artifactId>
			<version>${spring-version}</version>
		</dependency>
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-jms</artifactId>
			<version>${spring-version}</version>
		</dependency>
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-test</artifactId>
			<version>${spring-version}</version>
		</dependency>
		<dependency>
			<groupId>javax.annotation</groupId>
			<artifactId>jsr250-api</artifactId>
			<version>1.0</version>
		</dependency>
		<dependency>
			<groupId>org.apache.activemq</groupId>
			<artifactId>activemq-core</artifactId>
			<version>5.7.0</version>
		</dependency>
		
		<dependency>
			<groupId>org.apache.activemq</groupId>
			<artifactId>activemq-pool</artifactId>
			<version>5.13.0</version>
		</dependency>
		
	</dependencies>

3.spring-jms.xml配置文件

这里的配置类似于jdbc的配置，需要注意的是消息监听器有几种实现方式，

<?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:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
	 http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
	 http://www.springframework.org/schema/aop
	 http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
     http://www.springframework.org/schema/tx 
     http://www.springframework.org/schema/tx/spring-tx-3.0.xsd
     http://www.springframework.org/schema/context 
     http://www.springframework.org/schema/context/spring-context-3.0.xsd
		http://cxf.apache.org/jaxws 
		http://cxf.apache.org/schemas/jaxws.xsd">

 	<context:component-scan base-package="com.besttone.jms" />  

	<!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供 -->
	<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
		<property name="brokerURL" value="tcp://localhost:61616" />
	</bean>

	<!-- 链接对象进行池化，提高效率，类似于jdbc中的 c3p0/druid等连接池 -->
	<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
		<property name="connectionFactory" ref="targetConnectionFactory" />
		<property name="maxConnections" value="10" />
	</bean>

	<bean id="connectionFactory"
		class="org.springframework.jms.connection.SingleConnectionFactory">
		<property name="targetConnectionFactory" ref="pooledConnectionFactory" />
	</bean>

	<!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
		<property name="connectionFactory" ref="connectionFactory" />
		<!-- 向jmsTemplate注入消息转换器，会自动帮我们转换为我们需要的对象 -->
		<property name="messageConverter" ref="objConverter"/>  
	</bean>

	<!--这个是队列目的地，点对点的 -->
<!-- 	<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg>
			<value>besttone</value>
		</constructor-arg>
	</bean> -->
	<!--这个是主题目的地，一对多的 -->
	<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
		<constructor-arg value="topic" />
	</bean>

	<!-- 默认的消息回复队列 -->
	<bean id="defaultResponseTopic" class="org.apache.activemq.command.ActiveMQTopic">
		<constructor-arg>
			<value>defaultResponseTopic</value>
		</constructor-arg>
	</bean>

	
	<!-- 类型转换器 -->  
	<!-- producer: object-> Message ;  consumer: Message -> object -->
	<bean id="objConverter" class="com.besttone.jms.converter.ObjectConverter"/>
	
		<!-- 消息监听适配器 -->
	<bean id="defaultResponseTopicListenerAdapter"
		class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
		<property name="delegate">
			<bean class="com.besttone.jms.listener.ConsumerListener" />
		</property>
		<!-- 在监听到该topic有消息时，便调用ConsumerListener内的receiveMessage -->
		<property name="defaultListenerMethod" value="receiveMessage" />
		
	</bean>
	
	<!-- 消息监听适配器对应的监听容器 -->
	<bean id="defaultResponseTopicListenerAdapterContainer"
		class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="connectionFactory" />
		<property name="destination" ref="defaultResponseTopic" />
		<property name="messageListener" ref="defaultResponseTopicListenerAdapter" />
	</bean>
	  
</beans>

4.MessageConverter转换类

这个类在producer端，将Object 转换为javax.jms.Message；在consumer端，可以将javax.jms.Message 转换为Object

public class ObjectConverter implements MessageConverter {

    public Object fromMessage(Message message) throws JMSException, MessageConversionException {
        ObjectMessage objMessage = (ObjectMessage) message;  
        return objMessage.getObject();
    }

    public Message toMessage(Object obj, Session session) throws JMSException, MessageConversionException {
        return session.createObjectMessage((Serializable) obj);  
    }

}

5.ProducerResponseListener类

在produce 端发送消息到Topic之后，如果consumer端消费之后，会返回一个消息，这个类就是处理producer接收回复类

public class ConsumerListener {

    public void receiveMessage(Object obj) {
        Msg msg = (Msg) obj;
        if(msg.getCode()==1){
            System.out.println("回复接收成功~");
        }else{
            System.out.println("回复接收失败!");
        }
        
    }
}

6.生产消息类

jmsTemplate.convertAndSend(destination, obj) // 这个方法会调用上面的转换类

@Component("producerService")
public class ProducerServiceImpl implements ProducerService {
    
    @Autowired
    private JmsTemplate jmsTemplate;  
  
    public void sendObjMsg(Destination destination, Serializable obj) {  
        jmsTemplate.convertAndSend(destination, obj);// 需要定义自己的MessageConverter 或者是使用默认的        
    } 

}

7.测试类：

public class ProducerConsumerTest {


    public static void main(String[] args) {
        ApplicationContext ac = new ClassPathXmlApplicationContext("classpath:spring-integration.xml");
        Destination destination =  (Destination) ac.getBean("topicDestination");
        ProducerService producerService =  (ProducerService) ac.getBean("producerService");
        for (int i=0; i<1; i++) {  
            Msg msg = new Msg();
            msg.setCode(0);
            msg.setMsg("start");
            producerService.sendObjMsg(destination,msg);  
        }  
    }
}

consumer工程和producer的类似，下面提供下载

下面这些是在学习MQ的时候，比较好的学习资料：

1.http://wenku.baidu.com/link?url=Ms3fDI7ji6tpQC_-AbiETX2xGj2NcBAVUg-pFVj5spc6gkdUBjFCVYSwfvX7mvnCTnvBjx0D1nr3M8EV5Dl9--aMamHBwh1QO8bgP7PFXKG

2.http://blog.csdn.net/jwdstef/article/details/17380471

3.http://haohaoxuexi.iteye.com/category/347649