spring整合activemq传递对象注意事项

ゝ一世哀愁。 2022-06-02 11:14 201阅读 0赞

初衷：两个项目分别部署在两台服务器上，项目之间以activemq传递object消息

拓朴结构：

服务器A：tomcat，项目A--生产者  
服务器B：tomcat、activemq、项目B--消费者

项目A主要代码

src/main/resources/ActiveMQ.xml

<?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">
 
 
 
 
 <amq:connectionFactory id="amqConnectionFactory"
 brokerURL="tcp://172.16.2.36:61616" userName="admin" password="admin" />
 
 
 <bean id="queueSender" class="com.abc.queue.QueueSender"></bean>
 
 
  
 <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
  
 <property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
 
 
 
 <property name="sessionCacheSize" value="100" />
 </bean>
 
 
 
 
 <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
  
 <constructor-arg ref="connectionFactory" />
 
 <property name="pubSubDomain" value="false" />
 </bean>
 
 
 <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
  
 <constructor-arg ref="connectionFactory" />
 
 <property name="pubSubDomain" value="true" />
 </bean>
 
 
 
 </beans>

com.abc.queue.QueueSender

package com.abc.queue;
 
 import java.io.Serializable;
 
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.Session;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.jms.core.JmsTemplate;
 import org.springframework.jms.core.MessageCreator;
 import org.springframework.stereotype.Component;
 
 /**
 * 
 * ClassName: QueueSender 
 * Function: 队列消息生产者，发送消息到队列 
 * Created on: 2017年12月5日 下午2:59:44 
 * 
 * @author: Wender(FengHaiBing) 
 * @version: 
 * @since: JDK 1.8
 */
 @Component("queueSender")
 public class QueueSender {
 	
 	Logger LOG = LoggerFactory.getLogger(QueueSender.class);
 	
 	/**
 	 * 因为还有一个jmsTopicTemplate，
 	 * 所以这里要用到@Qualifier，明确此实例变量对应的是哪一个JmsTemplate
 	 */
 	@Autowired
 	@Qualifier("jmsQueueTemplate")
 	private JmsTemplate jmsTemplate;//通过@Qualifier修饰符来注入对应的bean
 	
 	/**
 	 * 发送一条消息到指定的队列（目标）
 	 * @param queueName 队列名称
 	 * @param message 消息内容
 	 */
 	public void send(String queueName,final String message){
 		try {
 			jmsTemplate.send(queueName, new MessageCreator() {
 			@Override
 				public Message createMessage(Session session) throws JMSException {
 					return session.createObjectMessage((Serializable)message);
 				}
 			});
 			
 		} catch (Exception e) {
 			LOG.error("QueueSender.send() exception occured, sending msg " + message + " to " + queueName,e);
 		}
 	}
 	
 	
 }

项目B主要代码

src/main/resources/ActiveMQ.xml

<?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">
 
 	
 	
 	
 	<amq:connectionFactory id="amqConnectionFactory"
 		brokerURL="tcp://172.16.2.36:61616" userName="admin" password="admin" />
 
 	
 <bean id="queueVfReceiver" class="com.abc.queue.QueueVfReceiver"></bean>
 	
 	
 	 
 	<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
 		 
 		<property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
 		
 		
 		
 		<property name="sessionCacheSize" value="100" />
 	</bean>
 	
 	
 	
 	
 	<bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
 		 
 		<constructor-arg ref="connectionFactory" />
 		
 		<property name="pubSubDomain" value="false" />
 	</bean>
 	
 	
 	<bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
 		  
 		<constructor-arg ref="connectionFactory" />
 		
 		<property name="pubSubDomain" value="true" />
 	</bean>
 	
 	
 
 	
 	
 	
 	
 	<jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
 		<jms:listener destination="queue.lv.img" ref="queueLvReceiver"/>
 	</jms:listener-container>
 	
 	
 </beans>

com.abc.queue.QueueVfReceiver

接收到对象后，需要用ActiveMQObjectMessage，做一次类型转换，将message的object转换为需要用的类型

package com.abc.queue;
 
 
 import javax.jms.Message;
 import javax.jms.MessageListener;
 import javax.jms.ObjectMessage;
 
 import org.apache.activemq.command.ActiveMQObjectMessage;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import com.alibaba.druid.util.StringUtils;
 import com.abc.facade.entity.VfInfo;
 
 /**
  * 
  * ClassName: QueueReceiver  
  * Function:  ADD FUNCTION.  
  * Created on: 2017年12月5日 下午1:56:46  
  *  
  * @author: Wender(FengHaiBing)  
  * @version:   
  * @since: JDK 1.8
  */
 @Component
 public class QueueVfReceiver implements MessageListener {
     
 //    @Override
     public void onMessage(Message message) {
         try {
             if (null == message) {
                 return ;
             }
             ActiveMQObjectMessage aMsg = (ActiveMQObjectMessage)message;
             VfInfo vfInfo = (VfInfo)aMsg.getObject();
             if(StringUtils.isEmpty(vfInfo.getId())) {
                 //不处理无效数据
                 return ;
             }            
             ....
  } catch (Exception e) {
             e.printStackTrace();
         }
     }
 }

activemq配置

将需要传递的对象所在的包，写入到变量ACTIVEMQ\_OPTS

if [ -z "$ACTIVEMQ_OPTS" ] ; then
    #    ACTIVEMQ_OPTS="$ACTIVEMQ_OPTS_MEMORY -Djava.util.logging.config.file=logging.properties -Djava.security.auth.login.config=$ACTIVEMQ_CONF/login.config"
        ACTIVEMQ_OPTS="$ACTIVEMQ_OPTS_MEMORY -Djava.util.logging.config.file=logging.properties -Djava.security.auth.login.config=$ACTIVEMQ_CONF/login.config -Dorg.apache.activemq.SERIALIZABLE_PACKAGES=java.lang,javax.security,java.util,org.apache.activemq,com.abc.facade.entity"
    fi