spring整合activeMQ之动态注入

83 阅读 0 评论 55 点赞

我是靠谱客的博主优美咖啡豆，最近开发中收集的这篇文章主要介绍spring整合activeMQ之动态注入，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

1.环境
Spring采用4.1.3版本，ActiveMQ使用5.9.1版本，本文不介绍Spring集成ActiveMQ来发送和接收JMS消息的细节。

2.参考
Spring整合JMS(一)——基于ActiveMQ实现系列。

3.基础配置
activemq.xml

<?xml version="1.0" encoding="UTF-8"?>
<activeMQ>
<brokerUrl>tcp://192.168.0.100:9300</brokerUrl>
<thisCode>central</thisCode>
<destination>
<queue code="reply" name="reply.Queue" remark="中心端接收回复消息的队列" />
<queue code="one" name="one.Queue" remark="一号测试队列" />
<queue code="two" name="two.Queue" remark="二号测试队列" />
<queue code="three" name="three.Queue" remark="三号测试队列" />
<queue code="four" name="four.Queue" remark="四号测试队列" />
</destination>
<listeners>
<queueListener code="reply" remark="中心端回复队列监听器"/>
</listeners>
</activeMQ>

applicationContext.xml 加入注解支持

<context:annotation-config/>
<bean class="org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor"/>
<!-- 打开Spring组件自动扫面，并配置要扫描的基本包 -->
<context:component-scan base-package="com.**.messaging.configuration"></context:component-scan>

4、动态注入队列

ActiveMQConfiguration.java在自己项目中的com.**.messaging.configuration包中，这里动态向spring容器注入队列及监听器，ActiveMQConfig是使用XStream将activemq.xml转化后的类，方便使用，具体找度娘，就不多说了
ApplicationContextFactory.getApplicationContext()是自定义的工厂方法，获取web环境上下文ApplicationContext，可通过实现ApplicationContextAware接口获取ApplicationContext，
QueueMsgListener和TopicMsgListener是实现spring的MessageListener接口的监听器，这里是因为要区分消息来自queue或topic所以分别做了实现。

/**
* ActiveMQ 配置
*/
@Configuration
public class ActiveMQConfiguration{
private static final Logger LOGGER = LoggerFactory.getLogger(ActiveMQConfiguration.class);
private InputStream in = null;
public static ActiveMQConfig activeMQConfig;
@Autowired
private ActiveMQConnectionFactory jmsConnectionFactory;
public ActiveMQConfiguration(){
try {
//
String path = this.getClass().getResource("/").getPath();
in = this.getClass().getResource("/activemq.xml").openStream();
activeMQConfig = XmlUtil.toBeanFromFile(in, ActiveMQConfig.class);

} catch (IOException e) {
e.printStackTrace();
}
}
/**
*
* @return
*/
@Bean
public String dynamicLoading(){
DefaultListableBeanFactory acf = (DefaultListableBeanFactory) ApplicationContextFactory.getApplicationContext().getAutowireCapableBeanFactory();
// 通过BeanDefinitionBuilder创建bean定义
//注册队列
if(activeMQConfig.getDestination().getQueues()!=null)
for (Destination queue : activeMQConfig.getDestination().getQueues()) {
BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ActiveMQQueue.class);
beanDefinitionBuilder.addConstructorArgValue(queue.getName());
acf.registerBeanDefinition(queue.getCode(),beanDefinitionBuilder.getRawBeanDefinition());
}
//注册队列监听容器
if(activeMQConfig.getListeners().getQueueListener()!=null)
for (Listener queueListener : activeMQConfig.getListeners().getQueueListener()) {
BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(DefaultMessageListenerContainer.class);
beanDefinitionBuilder.addPropertyValue("connectionFactory", jmsConnectionFactory);
beanDefinitionBuilder.addPropertyValue("messageListener", acf.getBean("queueMsgListener",QueueMsgListener.class));
beanDefinitionBuilder.addPropertyValue("destination", acf.getBean(queueListener.getCode(),ActiveMQQueue.class));
acf.registerBeanDefinition(queueListener.getCode()+".Container",beanDefinitionBuilder.getRawBeanDefinition());
}
//注册all topic（这里固定了topic的名称："TOPIC.ALL"）
BeanDefinitionBuilder allBuilder = BeanDefinitionBuilder.genericBeanDefinition(ActiveMQTopic.class);
allBuilder.addConstructorArgValue("TOPIC.ALL");
acf.registerBeanDefinition("TOPIC.ALL",allBuilder.getRawBeanDefinition());
//注册all topic监听（这里固定了topic的监听容器名称：""TOPIC.ALL.Container""）
BeanDefinitionBuilder allListenerBuilder = BeanDefinitionBuilder.genericBeanDefinition(DefaultMessageListenerContainer.class);
allListenerBuilder.addPropertyValue("connectionFactory", jmsConnectionFactory);
allListenerBuilder.addPropertyValue("messageListener", acf.getBean("topicMsgListener",TopicMsgListener.class));
allListenerBuilder.addPropertyValue("destination", acf.getBean("TOPIC.ALL",ActiveMQTopic.class));
acf.registerBeanDefinition("TOPIC.ALL.Container",allListenerBuilder.getRawBeanDefinition());
// 创建一个连接工厂，用于程序连接到activemq代理。
@Bean(name = "jmsConnectionFactory")
public ActiveMQConnectionFactory jmsConnectionFactory(){
ActiveMQConnectionFactory amqf = new ActiveMQConnectionFactory(activeMQConfig.getBrokerUrl());
amqf.setCloseTimeout(10*1000);
amqf.setSendTimeout(10*1000);
return amqf;
}
// 创建服务端的jmsTemplate（一般情况下服务端和客户端的jmsTemplate可以设置相同，但为了效率因素，我们将其分开设置，服务端应尽量减少连接数量，所以使用singleConnectionFactory）。
@Bean(name = "jmsTemplate")
public JmsTemplate jmsTemplate(){
JmsTemplate jt = new JmsTemplate();
SingleConnectionFactory scf = new SingleConnectionFactory();
scf.setTargetConnectionFactory(jmsConnectionFactory);
scf.setReconnectOnException(true);//连接断开后重试
jt.setConnectionFactory(scf);
return jt;
}
//
@Bean(name = "Queue.Container")
//
public DefaultMessageListenerContainer queueListener(){
//
DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
//
container.setConnectionFactory(jmsConnectionFactory);
//
container.setDestination(applicationContext.getAutowireCapableBeanFactory().getBean("QUEUE.baiyun",ActiveMQQueue.class));
//
container.setMessageListener(applicationContext.getAutowireCapableBeanFactory().getBean("consumerMsgListener",ConsumerMsgListener.class));
//
return container;
//
}
//
//
@Bean(name = "Topic.Container")
//
public DefaultMessageListenerContainer topicListener(){
//
DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
//
container.setConnectionFactory(jmsConnectionFactory);
//
container.setDestination(applicationContext.getAutowireCapableBeanFactory().getBean("TOPIC.all",ActiveMQTopic.class));
//
container.setMessageListener(applicationContext.getAutowireCapableBeanFactory().getBean("consumerMsgListener",ConsumerMsgListener.class));
//
return container;
//
}
}

acf.getBean(“…”,Class)从容器中获取bean，第一个参数是bean的name（ @Bean(name=”**”）），要和注入是使用的相同；spring会自动率先注入先使用的bean。
注掉的部分是配置固个数的队列，为了使队列及监听可动态配置，所以使用动态注入的方式；
bean JmsTemplate 中的这行代码还是很必要的：

scf.setReconnectOnException(true);//连接断开后重试

测试过程中关闭了ActiveMQ服务器，会导致client端获取连接失败，报出如下异常，启动服务器后仍然报出，加上上面代码使连接重试解决问题，正常生产环境也避免不了网络中断的情况，所以加上上面这行代码很有必要

org.springframework.jms.UncategorizedJmsException: Uncategorized exception occured during JMS processing; nested exception is
...
org.apache.activemq.ConnectionFailedException: The JMS connection has failed: Connection reset
...
Caused by: org.apache.activemq.ConnectionFailedException: The JMS connection has failed: Connection reset

5、生产者 producer

//1、发送接口(JmsMsg是自定义的消息实体基类,包含必须的消息发送的目的地destination，以及自己封装的消息id等常规的信息)
public interface Sender<T extends JmsMsg>{
MsgReturn sendMessageBefore();
MsgReturn sendMessage(T t) ;
}
//2、实现获取发送目的地队列Destination和JmsTemplate，调用JmsTemplate.send(Destination,MessageCreator )发送消息，由于不同的消息类型（textMessage或bytesMessage等）的参数不尽相同，获取MessageCreator需要在子类中去实现getMessageCreator()
@Component
public abstract class BaseSender<T extends JmsMsg>
implements Sender<T>{
public static final Logger LOGGER = LoggerFactory.getLogger(BaseSender.class);
@Autowired
private JmsTemplate JmsTemplate ;
private ActiveMQDestination destination;
private T baseMesage;
@Override
public MsgReturn sendMessageBefore() {
LOGGER.debug("==sendBefore");
return new MsgReturn();
}
private MsgReturn
onMessage() {
try {
destination = (ActiveMQDestination)ApplicationContextFactory.getApplicationContext().getBean(baseMesage.getDestinationCode());
JmsTemplate =
(JmsTemplate) ApplicationContextFactory.getApplicationContext().getBean("jmsTemplate");
JmsTemplate.send(destination,getMessageCreator());
} catch (BeansException e) {
e.printStackTrace();
return new MsgReturn(false,baseMesage.getMsgId(),e.getMessage());
} catch (JmsException e) {
e.printStackTrace();
return new MsgReturn(false,baseMesage.getMsgId(),e.getMessage());
}
return new MsgReturn(baseMesage.getMsgId(),"success");
}
protected MessageCreator getMessageCreator(){
return null;
}
@Override
public final MsgReturn sendMessage(T t) {
this.baseMesage = t;
MsgReturn msgReturn = new MsgReturn();
msgReturn = sendMessageBefore();
if(msgReturn.getResult()){
LOGGER.info("==开始发送");
msgReturn = onMessage();
LOGGER.info("==结束发送："+msgReturn);
}
return msgReturn;
}
public T getBaseMesage() {
return baseMesage;
}
public void setBaseMesage(T baseMesage) {
this.baseMesage = baseMesage;
}
}
//3、这是发送bytesMessage的实现类，(BytesMsg继承自定义的JmsMsg),getMessageCreator()中将自定义的参数转换成标准的Message
@Component
public class BytesMsgSender extends BaseSender<BytesMsg> {
public static final Logger LOGGER = Logger.getLogger(BytesMsgSender.class);
@Override
public MsgReturn sendMessageBefore() {
super.sendMessageBefore();
//
int fileSize = 10485760;//10MB
//
if(getBaseMesage().getFileSize()>fileSize)
//
return new MsgReturn(false,null,"==发送的文件超过指定大小");
return new MsgReturn();
}
@Override
public MessageCreator getMessageCreator(){
return new MessageCreator() {
public Message createMessage(Session session)throws JMSException {
BytesMessage message = session.createBytesMessage();
message.writeBytes(getBaseMesage().getBytes());
if(getBaseMesage().getMsgId()==null)
getBaseMesage().setMsgId(UUID.randomUUID().toString());
message.setStringProperty(Global.MSG_ID, getBaseMesage().getMsgId());
//...

message.setStringProperty(Global.FILE_NAME, getBaseMesage().getFileName());
message.setStringProperty(Global.FILE_MD5, getBaseMesage().getFileMD5());
message.setIntProperty(Global.FILE_SIZE, getBaseMesage().getFileSize());
return message;
}
};
}
}

6、消费者 customer

//1、消息接受接口,注入的监听器在消息到达时会调用onMessage(),这里需要定义这个方法。
public interface Reciever extends MessageListener{
void onMessage();
void replyCallBack();//回调
}
//2、实现对消息的转化
@Component("baseReciever")
public abstract class BaseReciever implements Reciever{
public static final Logger LOGGER = Logger.getLogger(BaseReciever.class);
private Message message;
private JmsMsg jmsMsg;
public final void onMessage(Message message) {
this.message = message; //标准的message
initJmsMsg();
onMessage();
replyCallBack();
}
//将标准的message转化为自定义的message
public void initJmsMsg() {
try {
if(message instanceof TextMessage){
TextMessage msg = (TextMessage) message;
TextMsg textMsg = new TextMsg();
textMsg.setText(msg.getText());
textMsg.setMsgId(message.getStringProperty(Global.MSG_ID));
//...
this.jmsMsg = textMsg;
}else if(message instanceof BytesMessage){
BytesMessage msg = (BytesMessage) message;
byte[] bytes = new byte[new Long(msg.getBodyLength()).intValue()];
msg.readBytes(bytes);
BytesMsg bytesMsg = new BytesMsg();
bytesMsg.setBytes(bytes);
bytesMsg.setMsgId(message.getStringProperty(Global.MSG_ID));
//...
this.jmsMsg = bytesMsg;
}
} catch (JMSException e) {
e.printStackTrace();
} catch (Exception e){
e.printStackTrace();
}
}
public JmsMsg getJmsMsg() {
return jmsMsg;
}
public void setJmsMsg(JmsMsg jmsMsg) {
this.jmsMsg = jmsMsg;
}
@Override
public void replyCallBack() {
}
@Override
public void onMessage() {
}
}
//3、监听实现，因业务不同所以分别实现；onMessage()中实现具体业务，replyCallBack()可以在收到消息后发送回复消息等
@Component
public class QueueMsgListener extends BaseReciever {
@Override
public void onMessage() {
//getJmsMsg()
//...
}
@Override
public void replyCallBack() {
//getJmsMsg()
//...
}
}
@Component
public class TopicMsgListenerextends BaseReciever {
@Override
public void onMessage() {
//getJmsMsg()
//...
}
@Override
public void replyCallBack() {
//getJmsMsg()
//...
}
}