Spring整合ActiveMQ实现简单的消息队列
2017-12-12 10:48
731 查看
Spring整合ActiveMQ实现简单的消息队列
JMS简介
JMS(Java Message Service),Java消息服务。主要用于在生产者和消费者之间进行消息传送,生产者负责生产消息,消费者负责接收处理消息。在实际的业务中,生产者生成消息,发送消息,消息会保存到ActiveMQ队列中,对应的消费者会接收消息,完成特定的业务逻辑。消息的传递模型有两种,一种是点对点,一个生产者对应一个消费者;另一种是发布订阅模式,一个生产者对应多个消费者,Java应用和ActiveMQ之间使用JMS传送消息。Spring整合ActiveMQ
引入依赖
maven依赖如下[html] view plain copy print?<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.14.5</version>
</dependency>
<dependency>
<groupId>javax.annotation</groupId>
<artifactId>jsr250-api</artifactId>
<version>1.0</version>
</dependency>
<span style=“white-space:pre”> </span><dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-messaging</artifactId>&
4000
nbsp;
<version>{spring.version}</span><span class="tag"></</span><span class="tag-name">version</span><span class="tag">></span><span> </span></span></li><li class="alt"><span> <span class="tag"><</span><span class="tag-name">span</span><span> </span><span class="attribute">style</span><span>=</span><span class="attribute-value">"white-space:pre"</span><span class="tag">></span><span> </span><span class="tag"></</span><span class="tag-name">span</span><span class="tag">></span><span class="tag"></</span><span class="tag-name">dependency</span><span class="tag">></span><span> </span></span></li><li class=""><span> <span class="tag"><</span><span class="tag-name">span</span><span> </span><span class="attribute">style</span><span>=</span><span class="attribute-value">"white-space:pre"</span><span class="tag">></span><span> </span><span class="tag"></</span><span class="tag-name">span</span><span class="tag">></span><span class="tag"><</span><span class="tag-name">dependency</span><span class="tag">></span><span> </span></span></li><li class="alt"><span> <span class="tag"><</span><span class="tag-name">groupId</span><span class="tag">></span><span>org.springframework</span><span class="tag"></</span><span class="tag-name">groupId</span><span class="tag">></span><span> </span></span></li><li class=""><span> <span class="tag"><</span><span class="tag-name">artifactId</span><span class="tag">></span><span>spring-jms</span><span class="tag"></</span><span class="tag-name">artifactId</span><span class="tag">></span><span> </span></span></li><li class="alt"><span> <span class="tag"><</span><span class="tag-name">version</span><span class="tag">></span><span>{spring.version}</version>
<span style=“white-space:pre”> </span></dependency>
<dependency>
<groupId>org.apache.xbean</groupId>
<artifactId>xbean-spring</artifactId>
<version>4.5</version>
</dependency>
<dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-all</artifactId> <version>5.14.5</version> </dependency> <dependency> <groupId>javax.annotation</groupId> <artifactId>jsr250-api</artifactId> <version>1.0</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-messaging</artifactId> <version>${spring.version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-jms</artifactId> <version>${spring.version}</version> </dependency> <dependency> <groupId>org.apache.xbean</groupId> <artifactId>xbean-spring</artifactId> <version>4.5</version> </dependency>
Spring中配置ActiveMQ
完整配置文件为:[html] view plain copy print?<?xml version=“1.0” encoding=“UTF-8”?>
<beans xmlns=“http://www.springframework.org/schema/beans”
xmlns:context=“http://www.springframework.org/schema/context” xmlns:p=“http://www.springframework.org/schema/p”
xmlns:mvc=“http://www.springframework.org/schema/mvc” xmlns:tx=“http://www.springframework.org/schema/tx”
xmlns:aop=“http://www.springframework.org/schema/aop” xmlns:xsi=“http://www.w3.org/2001/XMLSchema-instance”
xmlns:jms=“http://www.springframework.org/schema/jms” xmlns:amq=“http://activemq.apache.org/schema/core”
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/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-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.14.5.xsd”>
<!–activemq连接工厂 –>
<amq:connectionFactory id=“activemqConnectionFactory” brokerURL=“tcp://127.0.0.1:61616” userName=“admin” password=“admin”></amq:connectionFactory>
<!–spring连接工厂,管理activemq连接工厂 –>
<bean id=“connectionFactory” class=“org.springframework.jms.connection.CachingConnectionFactory”>
<!–管理对象,对应activemq连接工厂 –>
<property name=“targetConnectionF
19954
actory” ref=“activemqConnectionFactory”></property>
<!–session缓存数量 –>
<property name=“sessionCacheSize” value=“100”/>
</bean>
<!–定义消息队列 –>
<bean id=“tttQueueDestination” class=“org.apache.activemq.command.ActiveMQQueue”>
<!–消息队列名称 –>
<constructor-arg>
<value>ttt.queue</value>
</constructor-arg>
</bean>
<bean id=“tttTopicDestination” class=“org.apache.activemq.command.ActiveMQTopic”>
<!–订阅主题名称 –>
<constructor-arg>
<value>ttt.topic</value>
</constructor-arg>
</bean>
<!–消息生产者 –>
<!–队列类型JmsTemplate –>
<bean id=“jmsQueueTemplate” class=“org.springframework.jms.core.JmsTemplate”>
<property name=“connectionFactory” ref=“connectionFactory”/>
<property name=“defaultDestination” ref=“tttQueueDestination”/>
<property name=“receiveTimeout” value=“10000”/>
<!–非pub/sub(发布/订阅)模式,即队列模式 –>
<property name=“pubSubDomain” value=“false”/>
</bean>
<!–Topic类型JmsTemplate –>
<bean id=“jmsTopicTemplate” class=“org.springframework.jms.core.JmsTemplate”>
<property name=“connectionFactory” ref=“connectionFactory”/>
<property name=“defaultDestination” ref=“tttTopicDestination”/>
<property name=“receiveTimeout” value=“10000”/>
<!–pub/sub(发布/订阅)模式 –>
<property name=“pubSubDomain” value=“true”/>
</bean>
<!–消息消费者–>
<jms:listener-container destination-type=“queue” container-type=“default” connection-factory=“connectionFactory” acknowledge=“auto”>
<!–队列监听器 –>
<jms:listener destination=“ttt.queue” ref=“queueListener”/>
</jms:listener-container>
<jms:listener-container destination-type=“topic” container-type=“default” connection-factory=“connectionFactory” acknowledge=“auto”>
<!–Topic监听器 –>
<jms:listener destination=“ttt.topic” ref=“topicListener”/>
</jms:listener-container>
</beans>
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p" xmlns:mvc="http://www.springframework.org/schema/mvc" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jms="http://www.springframework.org/schema/jms" xmlns:amq="http://activemq.apache.org/schema/core" 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/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-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.14.5.xsd"> <!--activemq连接工厂 --> <amq:connectionFactory id="activemqConnectionFactory" brokerURL="tcp://127.0.0.1:61616" userName="admin" password="admin"></amq:connectionFactory> <!--spring连接工厂,管理activemq连接工厂 --> <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <!--管理对象,对应activemq连接工厂 --> <property name="targetConnectionFactory" ref="activemqConnectionFactory"></property> <!--session缓存数量 --> <property name="sessionCacheSize" value="100"/> </bean> <!--定义消息队列 --> <bean id="tttQueueDestination" class="org.apache.activemq.command.ActiveMQQueue"> <!--消息队列名称 --> <constructor-arg> <value>ttt.queue</value> </constructor-arg> </bean> <bean id="tttTopicDestination" class="org.apache.activemq.command.ActiveMQTopic"> <!--订阅主题名称 --> <constructor-arg> <value>ttt.topic</value> </constructor-arg> </bean> <!--消息生产者 --> <!--队列类型JmsTemplate --> <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory" ref="connectionFactory"/> <property name="defaultDestination" ref="tttQueueDestination"/> <property name="receiveTimeout" value="10000"/> <!--非pub/sub(发布/订阅)模式,即队列模式 --> <property name="pubSubDomain" value="false"/> </bean> <!--Topic类型JmsTemplate --> <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory" ref="connectionFactory"/> <property name="defaultDestination" ref="tttTopicDestination"/> <property name="receiveTimeout" value="10000"/> <!--pub/sub(发布/订阅)模式 --> <property name="pubSubDomain" value="true"/> </bean> <!--消息消费者--> <jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto"> <!--队列监听器 --> <jms:listener destination="ttt.queue" ref="queueListener"/> </jms:listener-container> <jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto"> <!--Topic监听器 --> <jms:listener destination="ttt.topic" ref="topicListener"/> </jms:listener-container> </beans>
发送消息
发送Queue消息收到发送Queue的请求,创建队列消息发送对象,设置消息业务处理逻辑bean、数据,发送消息
[java] view plain copy print?@Override
public void sendQueueMessage() {
Random r = new Random();
Map<String, Object> map = new HashMap<String, Object>();
map.put(”message” + r.nextInt(100), r.nextInt(10000));
MessageSender sender = new MessageSender(“asyncProcessor.message”);
sender.sendMessage(map);
}
@Override public void sendQueueMessage() { Random r = new Random(); Map<String, Object> map = new HashMap<String, Object>(); map.put("message" + r.nextInt(100), r.nextInt(10000)); MessageSender sender = new MessageSender("asyncProcessor.message"); sender.sendMessage(map); }
队列消息发送对象
[java] view plain copy print?public class MessageSender implements Serializable { /** * */ private static final long serialVersionUID = -7611739730866180422L; private String bean; public MessageSender(String bean) { this.bean = bean; } public String getBean() { return bean; } public void sendMessage(Destination destination, Map<String, Object> map) { MessageHelper.sendMessage(destination, map, getBean()); } public void sendMessage(Map<String, Object> map) { MessageHelper.sendMessage(map, getBean()); } }
public class MessageSender implements Serializable { /** * */ private static final long serialVersionUID = -7611739730866180422L; private String bean; public MessageSender(String bean) { this.bean = bean; } public String getBean() { return bean; } public void sendMessage(Destination destination, Map<String, Object> map) { MessageHelper.sendMessage(destination, map, getBean()); } public void sendMessage(Map<String, Object> map) { MessageHelper.sendMessage(map, getBean()); } }
调用jmsTemplate发送队列消息
[java] view plain copy print?@Component
public class MessageHelper implements Serializable, InitializingBean {
/**
*
*/
private static final long serialVersionUID = 522520461829363235L;
@Autowired
private JmsTemplate jmsQueueTemplate;
private static JmsTemplate jmsTemplateProxy;
public JmsTemplate getJmsTemplate() {
return jmsQueueTemplate;
}
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsQueueTemplate = jmsTemplate;
}
public static void sendMessage(Destination destination, final Map<String, Object> map, String processor) {
jmsTemplateProxy.send(destination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
MapMessage message = session.createMapMessage();
message.setString(”processor”, processor);
message.setObject(”data”, map);
message.setJMSDestination(destination);
return message;
}
});
}
public static void sendMessage(Map<String, Object> map, String processor) {
jmsTemplateProxy.send(new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
MapMessage message = session.createMapMessage();
message.setObject(”processor”, processor);
message.setObject(”data”, map);
return message;
}
});
}
@Override
public void afterPropertiesSet() throws Exception {
jmsTemplateProxy = this.jmsQueueTemplate;
}
public static JmsTemplate getJmsTemplateProxy() {
return jmsTemplateProxy;
}
public static void setJmsTemplateProxy(JmsTemplate jmsTemplateProxy) {
MessageHelper.jmsTemplateProxy = jmsTemplateProxy;
}
}
@Component public class MessageHelper implements Serializable, InitializingBean { /** * */ private static final long serialVersionUID = 522520461829363235L; @Autowired private JmsTemplate jmsQueueTemplate; private static JmsTemplate jmsTemplateProxy; public JmsTemplate getJmsTemplate() { return jmsQueueTemplate; } public void setJmsTemplate(JmsTemplate jmsTemplate) { this.jmsQueueTemplate = jmsTemplate; } public static void sendMessage(Destination destination, final Map<String, Object> map, String processor) { jmsTemplateProxy.send(destination, new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { MapMessage message = session.createMapMessage(); message.setString("processor", processor); message.setObject("data", map); message.setJMSDestination(destination); return message; } }); } public static void sendMessage(Map<String, Object> map, String processor) { jmsTemplateProxy.send(new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { MapMessage message = session.createMapMessage(); message.setObject("processor", processor); message.setObject("data", map); return message; } }); } @Override public void afterPropertiesSet() throws Exception { jmsTemplateProxy = this.jmsQueueTemplate; } public static JmsTemplate getJmsTemplateProxy() { return jmsTemplateProxy; } public static void setJmsTemplateProxy(JmsTemplate jmsTemplateProxy) { MessageHelper.jmsTemplateProxy = jmsTemplateProxy; } }
发送Topic消息
收到发送Topic的请求,创建Topic消息发送对象,设置消息业务处理逻辑bean、数据,发送消息
[java] view plain copy print?@Override
public void sendTopicMessage() {
Random r = new Random();
Map<String, Object> map = new HashMap<String, Object>();
map.put(”topic” + r.nextInt(100), r.nextInt(10000));
TopicSender sender = new TopicSender(“asyncProcessor.topic”);
sender.sendTopic(map);
}
@Override public void sendTopicMessage() { Random r = new Random(); Map<String, Object> map = new HashMap<String, Object>(); map.put("topic" + r.nextInt(100), r.nextInt(10000)); TopicSender sender = new TopicSender("asyncProcessor.topic"); sender.sendTopic(map); }
Topic消息发送对象类
[java] view plain copy print?public class TopicSender implements Serializable { /** * */ private static final long serialVersionUID = -2422605328103788546L; private String processor; public TopicSender(String processor) { this.processor = processor; } public String getProcessor() { return processor; } public void setProcessor(String processor) { this.processor = processor; } public void sendTopic(Destination destination, Map<String, Object> map) { TopicHelper.send(destination, map, getProcessor()); } public void sendTopic(Map<String, Object> map) { TopicHelper.send(map, getProcessor()); } }
public class TopicSender implements Serializable { /** * */ private static final long serialVersionUID = -2422605328103788546L; private String processor; public TopicSender(String processor) { this.processor = processor; } public String getProcessor() { return processor; } public void setProcessor(String processor) { this.processor = processor; } public void sendTopic(Destination destination, Map<String, Object> map) { TopicHelper.send(destination, map, getProcessor()); } public void sendTopic(Map<String, Object> map) { TopicHelper.send(map, getProcessor()); } }
调用jmsTemplate发送Topic消息
[java] view plain copy print?@Component
public class TopicHelper implements Serializable, InitializingBean {
/**
*
*/
private static final long serialVersionUID = 2255365346654153127L;
@Autowired
private JmsTemplate jmsTopicTemplate;
private static JmsTemplate jmsTopicTemplateProxy;
public static JmsTemplate getJmsTemplateProxy() {
return jmsTopicTemplateProxy;
}
public static void send(Destination destination, final Map<String, Object> map, final String processor) {
jmsTopicTemplateProxy.send(destination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
MapMessage message = session.createMapMessage();
message.setString(”processor”, processor);
message.setObject(”data”, map);
return message;
}
});
}
public static void send(final Map<String, Object> map, final String processor) {
jmsTopicTemplateProxy.send(new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
MapMessage message = session.createMapMessage();
message.setString(”processor”, processor);
message.setObject(”data”, map);
return message;
}
});
}
public JmsTemplate getJmsTopicTemplate() {
return jmsTopicTemplate;
}
public void setJmsTopicTemplate(JmsTemplate jmsTopicTemplate) {
this.jmsTopicTemplate = jmsTopicTemplate;
}
@Override
public void afterPropertiesSet() throws Exception {
jmsTopicTemplateProxy = getJmsTopicTemplate();
}
}
@Component public class TopicHelper implements Serializable, InitializingBean { /** * */ private static final long serialVersionUID = 2255365346654153127L; @Autowired private JmsTemplate jmsTopicTemplate; private static JmsTemplate jmsTopicTemplateProxy; public static JmsTemplate getJmsTemplateProxy() { return jmsTopicTemplateProxy; } public static void send(Destination destination, final Map<String, Object> map, final String processor) { jmsTopicTemplateProxy.send(destination, new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { MapMessage message = session.createMapMessage(); message.setString("processor", processor); message.setObject("data", map); return message; } }); } public static void send(final Map<String, Object> map, final String processor) { jmsTopicTemplateProxy.send(new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { MapMessage message = session.createMapMessage(); message.setString("processor", processor); message.setObject("data", map); return message; } }); } public JmsTemplate getJmsTopicTemplate() { return jmsTopicTemplate; } public void setJmsTopicTemplate(JmsTemplate jmsTopicTemplate) { this.jmsTopicTemplate = jmsTopicTemplate; } @Override public void afterPropertiesSet() throws Exception { jmsTopicTemplateProxy = getJmsTopicTemplate(); } }
监听消息
监听队列消息配置监听
[html] view plain copy print?<jms:listener-container destination-type=“queue” container-type=“default” connection-factory=“connectionFactory” acknowledge=“auto”>
<!–队列监听器 –>
<jms:listener destination=“ttt.queue” ref=“queueListener”/>
</jms:listener-container>
<jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto"> <!--队列监听器 --> <jms:listener destination="ttt.queue" ref="queueListener"/> </jms:listener-container>
监听实现
[java] view plain copy print?@Component
public class QueueListener implements MessageListener {
private static final Logger log = LoggerFactory.getLogger(MessageListener.class);
@SuppressWarnings(“unchecked”)
@Override
public void onMessage(Message message) {
MapMessage msg = (MapMessage) message;
try {
String processorName = msg.getString(”processor”);
Map<String, Object> param = (Map<String, Object>) msg.getObject(”data”);
IAsyncProcessor processor = (IAsyncProcessor) Context.getApplicationContext().getBean(processorName);
processor.process(param);
} catch (JMSException e) {
log.error(”QueueListener on message error|”, e);
}
}
}
@Component public class QueueListener implements MessageListener { private static final Logger log = LoggerFactory.getLogger(MessageListener.class); @SuppressWarnings("unchecked") @Override public void onMessage(Message message) { MapMessage msg = (MapMessage) message; try { String processorName = msg.getString("processor"); Map<String, Object> param = (Map<String, Object>) msg.getObject("data"); IAsyncProcessor processor = (IAsyncProcessor) Context.getApplicationContext().getBean(processorName); processor.process(param); } catch (JMSException e) { log.error("QueueListener on message error|", e); } } }
监听Topic消息
配置监听
[html] view plain copy print?<jms:listener-container destination-type=“topic” container-type=“default” connection-factory=“connectionFactory” acknowledge=“auto”>
<!–Topic监听器 –>
<jms:listener destination=“ttt.topic” ref=“topicListener”/>
</jms:listener-container>
<jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto"> <!--Topic监听器 --> <jms:listener destination="ttt.topic" ref="topicListener"/> </jms:listener-container>
监听实现
[java] view plain copy print?@Component
public class TopicListener implements MessageListener {
private static final Logger log = LoggerFactory.getLogger(MessageListener.class);
@SuppressWarnings(“unchecked”)
@Override
public void onMessage(Message message) {
MapMessage msg = (MapMessage) message;
try {
String processorName = msg.getString(”processor”);
Map<String, Object> param = (Map<String, Object>) msg.getObject(”data”);
IAsyncProcessor processor = (IAsyncProcessor) Context.getApplicationContext().getBean(processorName);
processor.process(param);
} catch (JMSException e) {
log.error(”TopicListener on message error|”, e);
}
}
}
@Component public class TopicListener implements MessageListener { private static final Logger log = LoggerFactory.getLogger(MessageListener.class); @SuppressWarnings("unchecked") @Override public void onMessage(Message message) { MapMessage msg = (MapMessage) message; try { String processorName = msg.getString("processor"); Map<String, Object> param = (Map<String, Object>) msg.getObject("data"); IAsyncProcessor processor = (IAsyncProcessor) Context.getApplicationContext().getBean(processorName); processor.process(param); } catch (JMSException e) { log.error("TopicListener on message error|", e); } } }
业务处理
业务逻辑处理bean实现
接口
[java] view plain copy print?public interface IAsyncProcessor { /** * 逻辑处理 * * @param map */ public void process(Map<String, Object> map); }
public interface IAsyncProcessor { /** * 逻辑处理 * * @param map */ public void process(Map<String, Object> map); }
实现
[java] view plain copy print?@Named(“asyncProcessor.message”)
public class MessageAsyncProcessor implements IAsyncProcessor {
@Override
public void process(Map<String, Object> map) {
for (Entry<String, Object> e : map.entrySet()) {
System.out.println(”message:” + e.getKey() + “,” + e.getValue());
}
}
}
@Named("asyncProcessor.message") public class MessageAsyncProcessor implements IAsyncProcessor { @Override public void process(Map<String, Object> map) { for (Entry<String, Object> e : map.entrySet()) { System.out.println("message:" + e.getKey() + "," + e.getValue()); } } }[java] view plain copy print?@Named(“asyncProcessor.topic”)
public class TopicAsyncProcessor implements IAsyncProcessor {
@Override
public void process(Map<String, Object> map) {
for (Entry<String, Object> e : map.entrySet()) {
System.out.println(”topic:” + e.getKey() + “,” + e.getValue());
}
}
}
@Named("asyncProcessor.topic") public class TopicAsyncProcessor implements IAsyncProcessor { @Override public void process(Map<String, Object> map) { for (Entry<String, Object> e : map.entrySet()) { System.out.println("topic:" + e.getKey() + "," + e.getValue()); } } }
获取业务逻辑处理bean的context代码
[java] view plain copy print?@Component public class Context implements ApplicationContextAware { private static ApplicationContext context; @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { context = applicationContext; } public static ApplicationContext getApplicationContext() { return context; } }
@Component public class Context implements ApplicationContextAware { private static ApplicationContext context; @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { context = applicationContext; } public static ApplicationContext getApplicationContext() { return context; } }
碰到问题
引入Activemq-all新版本jar包后,activemq中的Spring部分和Spring重复,会导致spring启动报错,只需要包activemq中的spring部分删除即可以正常运行。
(完)。
转自:http://blog.csdn.net/weihao_/article/details/74454736
相关文章推荐
- Spring整合ActiveMQ实现简单的消息队列
- spring boot整合activeMQ,实现ptp和topic两者消息模式
- SpringMVC4整合ActiveMQ实现消息队列
- spring整合消息中间件ActiveMQ简单使用
- 详解Java消息队列-Spring整合ActiveMq
- ActiveMQ整合Spring实现消息机制二(消息接收端配置)
- ActiveMQ消息队列和spring进行整合实例
- springboot 集成activeMQ实现消息队列和双向队列
- ActiveMQ实战(六)--Spring整合ActiveMQ实现队列和主题发布订阅通信
- spring整合apache activemq实现消息发送的三种方式代码配置实例
- Java消息队列-Spring整合ActiveMq
- 第二篇:spring+activeMQ实现消息队列简单demo
- Spring和ActiveMQ集成实现队列消息以及PUB/SUB模型
- spring-jms(activemq实现)使用queue发送消息简单例子
- 消息中间件之ActiveMQ整合Spring实现邮箱发送(四)
- 简单实现Java消息队列之activemq
- spring整合JMS实现同步收发消息(基于ActiveMQ的实现)
- Java消息队列-Spring整合ActiveMq
- Java消息队列-Spring整合ActiveMq
- spring整合activemq消息队列之点对点模式