spring与activeMQ（嵌入式启动activemq）

dwj147258

浏览: 191772 次
性别:
来自: 深圳

最近访客更多访客>>

q412378653

youyuezym

zlrzlm7454

skypiggy

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

spring框架

spring activemq 嵌入 java

一、准备工作

一般的我们发送jms消息都需要启动activemq来当做服务器，消息发送到activemq后，在通过监听activemq的消息来接收到消息，activemq下载地址http://activemq.apache.org/download.html，然后通过activemq.bat来启动，在这里我将通过一个例子来慢慢的讲解spring嵌入式启动ActiveMQ以及ActiveMQ监听器，发送订阅和点对点消息等，用到的是springMVC，这样我们就不需要启动activemq服务器就能够发送消息了，需要导入springmvc相关的jar包和activemqall包activemqjar包可以在下载的文件夹中找到。

二、从配置文件说起

首先是web项目的起点说起，web.xml：

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://java.sun.com/xml/ns/javaee" xmlns:web="http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd" xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd" id="WebApp_ID" version="3.0">
  <display-name>StudentManager</display-name>
  <welcome-file-list>
    <welcome-file>index.html</welcome-file>
  </welcome-file-list>
  <servlet>
		<servlet-name>websocket</servlet-name>
		<servlet-class>main.java.test.websocket.WebSocketInitServlet</servlet-class>
	</servlet>
	<servlet-mapping>
		<servlet-name>websocket</servlet-name>
		<url-pattern>/websocket.ws</url-pattern>
	</servlet-mapping>
	<!-- <context-param>
		<param-name>contextConfigLocation</param-name>
		<param-value>classpath:main/resource/ActiveMQ.xml</param-value>
	</context-param> -->
  	<servlet>  
     <servlet-name>spring</servlet-name>  
      <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
       <init-param>  
            <param-name>contextConfigLocation</param-name>  
            <param-value>classpath:main/resource/spring-servlet.xml</param-value>  
        </init-param>  
      <load-on-startup>1</load-on-startup>  
  	</servlet>
 	 <servlet-mapping>  
      <servlet-name>spring</servlet-name>  
      <url-pattern>/*</url-pattern>  
  	</servlet-mapping>  
  	<context-param>  
      <param-name>contextConfigLocation</param-name>  
      <param-value>classpath:main/resource/applicationContext.xml</param-value>  
  </context-param> 
  
</web-app>

这里面大家应该都能看懂，配置文件定义了两个servlet一个websocket一个spring，第一个大家可以不用关心，springservlet是springMVC的核心，这里关注的是spring的配置文件，applicationContext.xml这里先贴出配置文件，

<?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:jms="http://www.springframework.org/schema/jms"
          xmlns:amq="http://activemq.apache.org/schema/core"
        xsi:schemaLocation="  
        	  http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.8.0.xsd
              http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd  
              http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd
              http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context-3.0.xsd">  
  	<import resource="myActiveMQDemo.xml"/>
  </beans>

这里就是引入了一个activemq相关的配置文件，myActiveMQDemo.xml就是配置spring中的ActiveMQ如下：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:amq="http://activemq.apache.org/schema/core"
	xmlns:jms="http://www.springframework.org/schema/jms"
	xsi:schemaLocation="http://www.springframework.org/schema/beans   
        http://www.springframework.org/schema/beans/spring-beans-4.0.xsd   
        http://www.springframework.org/schema/context   
        http://www.springframework.org/schema/context/spring-context-4.0.xsd
        http://www.springframework.org/schema/jms
        http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
        http://activemq.apache.org/schema/core
        http://activemq.apache.org/schema/core/activemq-core-5.8.0.xsd">
        
     <bean id="broker" class="org.apache.activemq.xbean.BrokerFactoryBean"> 
   	 <property name="config" value="classpath:main/resource/ActiveMQConfig.xml" /> 
   	 <property name="start"  value="true" /> 
	</bean>  
	 <bean id="myamqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
   		 <property name="brokerURL" value="tcp://localhost:61616"/>  
   		 <property name="trustedPackages">
        <list>
            <value>main.java</value>
        </list>
    </property> 
	</bean>
     <bean id="myconnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
        <property name="targetConnectionFactory" ref="myamqConnectionFactory"></property>
        <property name="sessionCacheSize" value="100" />
    </bean>
    

    <bean id="myjmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
        <constructor-arg ref="myconnectionFactory" />
        <property name="pubSubDomain" value="false" />
    </bean>

    <bean id="myjmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
        <constructor-arg ref="myconnectionFactory" />
        <property name="pubSubDomain" value="true" />
    </bean>
    <bean id="mymsgQueueSender" class="main.java.test.jms.mydemo.Service.MsgQueueSenderImp">
    	<property name="jmsTemplate" ref="myjmsQueueTemplate"/>
    </bean>
	 <bean id="mymsgTopicSender" class="main.java.test.jms.mydemo.Service.MsgTopicSenderImp">
    	<property name="jmsTemplate" ref="myjmsTopicTemplate"/>
    </bean>
	<bean id="myqueueDestination" class="org.apache.activemq.command.ActiveMQQueue">  
  		  <constructor-arg>  
     		   <value>msg_topic_two</value>  
  		  </constructor-arg>  
	</bean>  
	<!--这个是主题目的地，一对多的-->  
	<bean id="mytopicDestination" class="org.apache.activemq.command.ActiveMQTopic">  
 		   <constructor-arg value="msg_topic_one"/>  
	</bean>
	<bean id="mytopicDestination2" class="org.apache.activemq.command.ActiveMQTopic">  
 		   <constructor-arg value="msg_topic_two"/>  
	</bean>
	
	<!-- 消息监听器 -->  
	<bean id="mytopicConsumer" class="main.java.test.jms.mydemo.consumer.TopicConsumer">
		<property name="destination" ref="mytopicDestination"/>
	</bean>
	<bean id="mytopicConsumer2" class="main.java.test.jms.mydemo.consumer.TopicConsumer2">
		<property name="destination" ref="mytopicDestination"/>
	</bean>       
	
    <bean id="myqueueDestinationMessageListenerAdapter" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">  
     <property name="delegate">  
        <bean class="main.java.test.jms.mydemo.consumer.TopicConsumer"/> 
    </property>
     <property name="defaultListenerMethod" value="onMessage"/> 
    </bean>  
    <bean id="myjMSReceiverQueueListenerAdapter" class=" org.springframework.jms.listener.DefaultMessageListenerContainer">    
          <property name="connectionFactory" ref="myconnectionFactory" />    
        <property name="destination" ref="mytopicDestination" />    
        <property name="messageListener" ref="myqueueDestinationMessageListenerAdapter" /> 
    </bean> 
    
     <bean id="myqueueDestinationMessageListenerAdapter2" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">  
     <property name="delegate">  
        <bean class="main.java.test.jms.mydemo.consumer.TopicConsumer2"/> 
    </property>
     <property name="defaultListenerMethod" value="onMessage"/> 
    </bean>  
    <bean id="myjMSReceiverQueueListenerAdapter2" class=" org.springframework.jms.listener.DefaultMessageListenerContainer">    
          <property name="connectionFactory" ref="myconnectionFactory" />    
        <property name="destination" ref="mytopicDestination" />    
        <property name="messageListener" ref="myqueueDestinationMessageListenerAdapter2" /> 
    </bean>
    
     <bean id="myqueueDestinationMessageListenerAdapter3" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">  
     <property name="delegate">  
        <bean class="main.java.test.jms.mydemo.consumer.QueueConsumer"/> 
    </property>
     <property name="defaultListenerMethod" value="onMessage"/> 
    </bean>  
    <bean id="myjMSReceiverQueueListenerAdapter3" class=" org.springframework.jms.listener.DefaultMessageListenerContainer">    
          <property name="connectionFactory" ref="myconnectionFactory" />    
        <property name="destination" ref="myqueueDestination" />    
        <property name="messageListener" ref="myqueueDestinationMessageListenerAdapter3" /> 
    </bean>
    <!-- 方式二  end-->
</beans>

这个配置文件首先配置了一个broker，这是ActiveMQ服务的关键bean，有了他我们就可以不用启动ActiveMQ服务了，而ActiveMQConfig配置文件的内容如下：

<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:amq="http://activemq.apache.org/schema/core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
  http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
    <bean
        class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer" />
    <broker useJmx="false" persistent="false"
        xmlns="http://activemq.apache.org/schema/core">
        <transportConnectors>
            <transportConnector uri="tcp://localhost:61616"/>
        </transportConnectors>
    </broker>
</beans>

到这里，配置文件就结束了，看到这里的同学可能觉得有点绕，这里我解释一下，首先进入spring的配置文件，applicationContext.xml这个配置文件配置了spring中配置ActiveMQ的全部配置信息的文件myActiveMQDemo.xml，myActiveMQDemo.xml这个配置文件就是配置所有的ActiveMQ的信息，首先，我们嵌入式启动ActiveMQ就需要定义出他的关键服务：broker，broker是服务我们jms消息的关键服务，classpath:main/resource/ActiveMQConfig.xml这个则是配置了broker服务的关键信息，其实就是配置了一个连接地址，接下来回到myActiveMQDemo.xml这个配置文件，往下走...

<bean id="myamqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
   		 <property name="brokerURL" value="tcp://localhost:61616"/>  
   		 <property name="trustedPackages">
        <list>
            <value>main.java</value>
        </list>
    </property> 
	</bean>
     <bean id="myconnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
        <property name="targetConnectionFactory" ref="myamqConnectionFactory"></property>
        <property name="sessionCacheSize" value="100" />
    </bean>

这两个bean是配置ActiveMQ的连接信息，第一个是配置连接工厂，真正连接的是myconnectionFactory这个bean，我们使用的是CachingConnectionFactory，接下来就是配置发送消息的spring模板bean，因为虽然我们配置了连接工厂，就如同spring的数据库连接模板一样，他为你写好了很多的样板代码，我们只需要调用模板中的发送方法就可以发送消息了，模板配置：

<bean id="myjmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
        <constructor-arg ref="myconnectionFactory" />
        <property name="pubSubDomain" value="true" />
    </bean>
    <bean id="mymsgQueueSender" class="main.java.test.jms.mydemo.Service.MsgQueueSenderImp">
    	<property name="jmsTemplate" ref="myjmsQueueTemplate"/>
    </bean>

这两个template可以说是一样的，pubSubDomain属性时配置发送订阅消息还是发送点对点消息，第一个bean就表示这是一个可以发送订阅消息的bean，只要把它注入到服务类中，调用他的方法就可以发送订阅消息，第二个bean是配置点对点的template。

<bean id="mymsgQueueSender" class="main.java.test.jms.mydemo.Service.MsgQueueSenderImp">
    	<property name="jmsTemplate" ref="myjmsQueueTemplate"/>
    </bean>
	 <bean id="mymsgTopicSender" class="main.java.test.jms.mydemo.Service.MsgTopicSenderImp">
    	<property name="jmsTemplate" ref="myjmsTopicTemplate"/>
    </bean>

这里就是将template注入到服务类中，后面会贴出服务类的java代码，接下来就是配置目的地：

<bean id="myqueueDestination" class="org.apache.activemq.command.ActiveMQQueue">  
  		  <constructor-arg>  
     		   <value>msg_topic_two</value>  
  		  </constructor-arg>  
	</bean>  
	<!--这个是主题目的地，一对多的-->  
	<bean id="mytopicDestination" class="org.apache.activemq.command.ActiveMQTopic">  
 		   <constructor-arg value="msg_topic_one"/>  
	</bean>
	<bean id="mytopicDestination2" class="org.apache.activemq.command.ActiveMQTopic">  
 		   <constructor-arg value="msg_topic_two"/>  
	</bean>

配置了三个目的地，分别是点对点消息目的地，msg_topic_two ，订阅消息目的地：msg_topic_one和msg_topic_two接下来就是配置消息监听器，消息监听器有三种，可以参考;

http://dwj147258.iteye.com/blog/2330295,这里我们是用的第三种

先贴出xml代码：

<bean id="myqueueDestinationMessageListenerAdapter" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">  
     <property name="delegate">  
        <bean class="main.java.test.jms.mydemo.consumer.TopicConsumer"/> 
    </property>
     <property name="defaultListenerMethod" value="onMessage"/> 
 <property name="defaultResponseDestination" ref="defaultResponseQueue"/> 
    </bean>  
    <bean id="myjMSReceiverQueueListenerAdapter" class=" org.springframework.jms.listener.DefaultMessageListenerContainer">    
          <property name="connectionFactory" ref="myconnectionFactory" />    
        <property name="destination" ref="mytopicDestination" />    
        <property name="messageListener" ref="myqueueDestinationMessageListenerAdapter" /> 
    </bean> 
    
     <bean id="myqueueDestinationMessageListenerAdapter2" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">  
     <property name="delegate">  
        <bean class="main.java.test.jms.mydemo.consumer.TopicConsumer2"/> 
    </property>
     <property name="defaultListenerMethod" value="onMessage"/> 
    </bean>  
    <bean id="myjMSReceiverQueueListenerAdapter2" class=" org.springframework.jms.listener.DefaultMessageListenerContainer">    
          <property name="connectionFactory" ref="myconnectionFactory" />    
        <property name="destination" ref="mytopicDestination" />    
        <property name="messageListener" ref="myqueueDestinationMessageListenerAdapter2" /> 
    </bean>
    
     <bean id="myqueueDestinationMessageListenerAdapter3" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">  
     <property name="delegate">  
        <bean class="main.java.test.jms.mydemo.consumer.QueueConsumer"/> 
    </property>
     <property name="defaultListenerMethod" value="onMessage"/> 
    </bean>  
    <bean id="myjMSReceiverQueueListenerAdapter3" class=" org.springframework.jms.listener.DefaultMessageListenerContainer">    
          <property name="connectionFactory" ref="myconnectionFactory" />    
        <property name="destination" ref="myqueueDestination" />    
        <property name="messageListener" ref="myqueueDestinationMessageListenerAdapter3" /> 
    </bean>

这是定义了三个消息监听器，拿一个出来解释就可以了，myqueueDestinationMessageListenerAdapter是定义了一个消息监听器，里面的属性delegate时指出消息监听类，defaultListenerMethod是指定接收消息的方法，defaultResponseDestination是指定回复消息的目的地，接下来就是配置消息监听器容器，里面的属性分别是配置连接工厂，目的地，以及消息监听器。到这里，配置问价那就已经全部结束。

三、前端和后台

在前端主要就是一个jsp页面：

<%@ page language="java" contentType="text/html; charset=utf-8"
    pageEncoding="utf-8"%>
<%
	String path = request.getContextPath();
	System.out.println(path);
	String basePath = request.getScheme() + "://"
			+ request.getServerName() + ":" + request.getServerPort()
			+ path + "/";
	System.out.println(basePath);
%>

<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">
<html>
<head>
<meta http-equiv="Content-Type" content="text/html; charset=utf-8">
<base href="<%=basePath%>">

<title>ActiveMQ Demo程序</title>

<meta http-equiv="pragma" content="no-cache">
<meta http-equiv="cache-control" content="no-cache">
<meta http-equiv="expires" content="0">
<script type="text/javascript" src="js/jquery.js"></script>
<style type="text/css">
.h1 {
	margin: 0 auto;
}

#producer{
	width: 48%;
 	border: 1px solid blue; 
	height: 80%;
	align:center;
	margin:0 auto;
}

body{
	text-align :center;
} 
div {
	text-align :center;
}
textarea{
	width:80%;
	height:100px;
	border:1px solid gray;
}
button{
	background-color: rgb(62, 156, 66);
	border: none;
	font-weight: bold;
	color: white;
	height:30px;
}
</style>
<script type="text/javascript">
	
	function send(controller){
		if($("#message").val()==""){
			$("#message").css("border","1px solid red");
			return;
		}else{
			$("#message").css("border","1px solid gray");
		}
		debugger
		$.ajax({
			type : "POST",
			url:"activemq/"+controller,
			dataType:'text', 
			data:{"message":$("#message").val()},
			success:function(data){
				if(data=="suc"){
					$("#status").html("<font color=green>发送成功</font>");
					setTimeout(clear,1000);
				}else{
					$("#status").html("<font color=red>"+data+"</font>");
					setTimeout(clear,5000);
				}
			},
			error:function(data){
				$("#status").html("<font color=red>ERROR:"+data["status"]+","+data["statusText"]+"</font>");
				setTimeout(clear,5000);
			}
			
		});
	}
	
	function clear(){
		$("#status").html("");
	}

</script>
</head>

<body>
	<h1>Hello ActiveMQ</h1>
	<div id="producer">
		<h2>Producer</h2>
		<textarea id="message"></textarea>
		<br>
		<button onclick="send('queueSender')">发送Queue消息</button>
		<button onclick="send('topicSender')">发送Topic消息</button>
		<br>
		<span id="status"></span>
	</div>
</body>
</html>

你需要引入你的jquery库，接着是控制器代码：

package main.java.test.jms.SpringActivemq.controller;



import javax.jms.Destination;

import main.java.bean.LoginUserInfo;
import main.java.test.jms.SpringAcrivemq2.ProducerServiceImpl;
import main.java.test.jms.SpringActivemq.mq.producer.queue.QueueSender;
import main.java.test.jms.SpringActivemq.mq.producer.topic.TopicSender;
import main.java.test.jms.mydemo.Service.MsgQueueSenderImp;
import main.java.test.jms.mydemo.Service.MsgTopicSenderImp;
import main.java.test.jms.mydemo.consumer.ConstUtil;
import main.java.util.BeanUtil;

import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;


/**
 * 
 * @author liang
 * @description controller娴嬭瘯
 */
@Controller
@RequestMapping("/activemq")
public class ActivemqController {
	
	
	/**
	 * 鍙戦�娑堟伅鍒伴槦鍒�
	 * Queue闃熷垪锛氫粎鏈変竴涓闃呰�浼氭敹鍒版秷鎭紝娑堟伅涓�棪琚鐞嗗氨涓嶄細瀛樺湪闃熷垪涓�
	 * @param message
	 * @return String
	 */
	@ResponseBody
	@RequestMapping("/queueSender")
	public String queueSender(@RequestParam("message")String message){
		String opt="";
		try {
//			queueSender.send("test.queue", message);
			MsgQueueSenderImp sender = (MsgQueueSenderImp)BeanUtil.getBeanByName("mymsgQueueSender");
			 LoginUserInfo user = new LoginUserInfo();
		        user.setId(2);
		        user.setPassWord("123");
		        user.setUserName("dengwei") ;
			sender.send(ConstUtil.MSG_TOPIC_TWO, user) ;
			opt = "suc";
		} catch (Exception e) {
			opt = e.getCause().toString();
		}
		return opt;
	}
	
	/**
	 * 鍙戦�娑堟伅鍒颁富棰�
	 * Topic涓婚 锛氭斁鍏ヤ竴涓秷鎭紝鎵�湁璁㈤槄鑰呴兘浼氭敹鍒�
	 * 杩欎釜鏄富棰樼洰鐨勫湴鏄竴瀵瑰鐨�
	 * @param message
	 * @return String
	 */
	@ResponseBody
	@RequestMapping("/topicSender")
	public String topicSender(@RequestParam("message")String message){
		String opt = "";
		try {
			System.out.println("fasong");
//			topicS.send("topic", message);
			MsgTopicSenderImp sender = (MsgTopicSenderImp)BeanUtil.getBeanByName("mymsgTopicSender");
			 LoginUserInfo user = new LoginUserInfo();
		        user.setId(2);
		        user.setPassWord("123");
		        user.setUserName("dengwei") ;
			sender.send(ConstUtil.MSG_TOPIC_ONE, user) ;
			opt = "suc";
		} catch (Exception e) {
			opt = e.getCause().toString();
		}
		return opt;
	}
	
}

实例类代码：

package main.java.bean;

import java.io.Serializable;

public class LoginUserInfo implements Serializable{

	private static final long serialVersionUID = -5668675616184265758L;

	private String userName ; 
	
	private String passWord ; 
	
	private String role ; 
	
	private int id  ;

	public String getUserName() {
		return userName;
	}

	public void setUserName(String userName) {
		this.userName = userName;
	}

	public String getPassWord() {
		return passWord;
	}

	public void setPassWord(String passWord) {
		this.passWord = passWord;
	}

	public String getRole() {
		return role;
	}

	public void setRole(String role) {
		this.role = role;
	}

	public int getId() {
		return id;
	}

	public void setId(int id) {
		this.id = id;
	}

	@Override
	public String toString() {
		return "LoginUserInfo [userName=" + userName + ", passWord=" + passWord
				+ ", role=" + role + ", id=" + id + "]";
	}
	
	
	
}

发送订阅消息接口：

package main.java.test.jms.mydemo.interfa;

import java.io.Serializable;

public interface MsgTopicSender {
	
	public void send(String topic , Serializable msg);
	
}

发送点对点消息接口：

package main.java.test.jms.mydemo.interfa;

import java.io.Serializable;

public interface MsgQueueSender {
	
	public void send(String queue , Serializable msg);
	
}

消费者接口：

package main.java.test.jms.mydemo.interfa;

import java.io.Serializable;

public interface Consumer {
	public void onMessage(Serializable msg);
	
	public String getDest() ;
}

目的地：

package main.java.test.jms.mydemo.consumer;

public class ConstUtil {
	public static String MSG_TOPIC_ONE = "msg_topic_one" ;
	
	public static String MSG_TOPIC_TWO ="msg_topic_two" ;
}

订阅消息发送类：

package main.java.test.jms.mydemo.Service;

import java.io.Serializable;

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

import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

import main.java.test.jms.mydemo.interfa.MsgTopicSender;

public class MsgTopicSenderImp implements MsgTopicSender{
	
	private JmsTemplate jmsTemplate ;
	
	@Override
	public void send(String topic,final Serializable msg) {
		jmsTemplate.send(topic, new MessageCreator() {  
            public Message createMessage(Session session) throws JMSException {  
            	ObjectMessage message = session.createObjectMessage();
            	message.setObject(msg);
                return message;  
            }  
        });
	}

	public JmsTemplate getJmsTemplate() {
		return jmsTemplate;
	}

	public void setJmsTemplate(JmsTemplate jmsTemplate) {
		this.jmsTemplate = jmsTemplate;
	}

	
}

点对点消息发送类：

package main.java.test.jms.mydemo.Service;

import java.io.Serializable;

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

import main.java.test.jms.mydemo.interfa.MsgQueueSender;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

public class MsgQueueSenderImp implements MsgQueueSender{
	
	private JmsTemplate jmsTemplate ;
	@Override
	public void send(String queue, final Serializable msg) {

		jmsTemplate.send(queue, new MessageCreator() {  
            public Message createMessage(Session session) throws JMSException {  
            	ObjectMessage message = session.createObjectMessage();
            	message.setObject(msg);
                return message;  
            }  
        });
		
	}
	public JmsTemplate getJmsTemplate() {
		return jmsTemplate;
	}
	public void setJmsTemplate(JmsTemplate jmsTemplate) {
		this.jmsTemplate = jmsTemplate;
	}
	
}

订阅消息消费者1：

package main.java.test.jms.mydemo.consumer;

import java.io.Serializable;

import javax.jms.Destination;

import main.java.bean.LoginUserInfo;
import main.java.test.jms.mydemo.interfa.Consumer;

public class TopicConsumer implements Consumer{
	private Destination destination ;
	
	@Override
	public void onMessage(Serializable msg) {
		System.out.println("Topicconsumer receive");
		if(msg instanceof LoginUserInfo){
			LoginUserInfo user = (LoginUserInfo)msg ;
			System.out.println(user);
		}
	}
	public void handleMessage(Serializable msg){
		System.out.println("handmessage");
	}
	@Override
	public String getDest() {
		return ConstUtil.MSG_TOPIC_ONE ;
	}

	public Destination getDestination() {
		return destination;
	}

	public void setDestination(Destination destination) {
		this.destination = destination;
	}

}

订阅消息消费者2：

package main.java.test.jms.mydemo.consumer;

import java.io.Serializable;

import javax.jms.Destination;

import main.java.bean.LoginUserInfo;
import main.java.test.jms.mydemo.interfa.Consumer;

public class TopicConsumer2 implements Consumer{
	private Destination destination ;
	@Override
	public void onMessage(Serializable msg) {
		System.out.println("Topicconsumer2 receive");
		if(msg instanceof LoginUserInfo){
			LoginUserInfo user = (LoginUserInfo)msg ;
			System.out.println(user);
		}
	}
	
	@Override
	public String getDest() {
		return ConstUtil.MSG_TOPIC_ONE ;
	}

	public Destination getDestination() {
		return destination;
	}

	public void setDestination(Destination destination) {
		this.destination = destination;
	}

}

点对点消息消费者：

package main.java.test.jms.mydemo.consumer;

import java.io.Serializable;

import main.java.bean.LoginUserInfo;
import main.java.test.jms.mydemo.interfa.Consumer;

public class QueueConsumer implements Consumer{

	@Override
	public void onMessage(Serializable msg) {
		System.out.println("QueueConsumer receive");
		if(msg instanceof LoginUserInfo){
			LoginUserInfo user = (LoginUserInfo)msg ;
			System.out.println(user);
		}
	}

	@Override
	public String getDest() {
		// TODO Auto-generated method stub
		return null;
	}

}

附上获取spring管理的bean的工具类：

package main.java.util;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.FileSystemXmlApplicationContext;
import org.springframework.web.context.support.WebApplicationObjectSupport;

public class BeanUtil extends WebApplicationObjectSupport{
	private static ApplicationContext ac = new FileSystemXmlApplicationContext("classpath:main/resource/applicationContext.xml");
	public static Object getBeanByName(String name ){
		return ac.getBean(name);
	}
	
}

四、一些问题

过程中遇到一些问题，

①发送一次订阅消息，接收到多次一样的，这是因为获取bean的工具类没有吧applicationContext静态，每调用一次就会加载一次导致创建了多个消息监听bean

②发送消息传输对象的时候总是报序列化异常，这是因为没有吧传输对象的包设置为白名单，具体配置是在配置连接工厂的时候，在brokerURL后跟一个属性，trustedPackages，然后列出白名单包列表。如下：

 <bean id="myamqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
   		 <property name="brokerURL" value="tcp://localhost:61616"/>  
   		 <property name="trustedPackages">
        <list>
            <value>main.java</value>
        </list>
    </property>

这里附上嵌入启动ActiveMQ的另一边值得推荐的博文http://www.cnblogs.com/dead-trap-ramble/archive/2013/11/23/3439060.html

0
顶

0
踩

分享到：

使用远程服务ＲＭＩ和ｉｎｖｏｋｅｒ | spring与activemq（三种消息监听方式）

2016-10-14 11:16
浏览 6860
评论(0)
分类:编程语言
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论