ActiveMQ整合spring

1、 activemq在spring中的配置  (生产者)

       说明:这里我使用的是topic方式  应用场景（添加了新的商品，可能需要同步索引库，缓存或者生成静态页面）应该有多个消费者消费同一个消息

<!-- 配置第三方jms connectionFactory 这里使用的是activemq-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<constructor-arg name="brokerURL" value="tcp://10.117.17.106:61616"/>
</bean>

<!-- 配置spring管理的jms connectionFactory-->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="targetConnectionFactory"/>
</bean>

<!-- 配置jmsTemplate -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"/>
</bean>

<!-- 配置topic -->
<bean id="itemAddTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg name="name" value="item-add-topic"/>
</bean>
2、 把jmsTemplate和需要的topic对象注入到类中成为类（我用的是controller，在添加商品提交事务过后进行发布消息）的属性

       下面是java代码片段:

@Autowired
private JmsTemplate jmsTemplate;
@Resource(name="itemAddTopic")
private Destination destination;
/**
* 增加新的商品
* @param item 商品的详细信息
* @param desc 商品的介绍
* @return
*/
@RequestMapping("/item/save")
@ResponseBody
public TaotaoResult addItem(TbItem item,String desc){
long itemId = ItemService.addItem(item,desc);
//发送到消息队列中
jmsTemplate.send(destination,new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
//发送itemId
TextMessage textMessage = session.createTextMessage(String.valueOf(itemId));
return textMessage;
}
});
return TaotaoResult.ok();
}
3、 activemq在spring中的配置 (消费者)

说明:这里订阅了生产者对应的商品添加话题 通过配置同样的destination

<!-- 配置第三方jms connectionFactory 这里使用的是activemq-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionF
4000
actory">
<constructor-arg name="brokerURL" value="tcp://10.117.17.106:61616"/>
</bean>

<!-- 配置spring管理的jms connectionFactory-->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="targetConnectionFactory"/>
</bean>

<!-- 配置queue -->
<bean id="test-queue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg name="name" value="test-queue"/>
</bean>

<!-- 配置topic -->
<bean id="itemAddTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg name="name" value="item-add-topic"/>
</bean>

<!-- 配置消息监听器 -->
<bean id="myMessageListener" class="com.taotao.search.listener.MyMessageListener"/>
<!-- 配置消息监听容器 -->
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="test-queue"/>
<property name="messageListener" ref="myMessageListener"></property>
</bean>

<!-- 配置添加商品消息监听器 -->
<bean id="itemAddMessageListener" class="com.taotao.search.listener.ItemAddMessageListener"/>
<!-- 配置消息监听容器 -->
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="itemAddTopic"/>
<property name="messageListener" ref="itemAddMessageListener"></property>
</bean>
4、 自定义messagelistener实现接口并实现方法

下面是java代码片段:

/**
* 这是同步商品添加索引的监听器
* @author xuguoqin
*
*/
public class ItemAddMessageListener implements MessageListener{

@Autowired
private SolrServer SolrServer;
@Autowired
private SearchItemMapper searchItemMapper;

@Override
public void onMessage(Message message) {
try {
//得到消息文本
TextMessage textMessage = (TextMessage)message;
//得到商品id
long itemId = Long.parseLong(textMessage.getText());
//得到searchItem对象
SearchItem searchItem = searchItemMapper.getSearchItemById(itemId);
//设置需要索引的字段
SolrInputDocument document = new SolrInputDocument();
document.addField("id", searchItem.getId());
document.addField("item_title", searchItem.getTitle());
document.addField("item_sell_point", searchItem.getSell_point());
document.addField("item_price", searchItem.getPrice());
document.addField("item_image", searchItem.getImage());
document.addField("item_category_name", searchItem.getCategory_name());
document.addField("item_desc", searchItem.getItem_desc());
//添加到文档索引中
SolrServer.add(document);
//提交
SolrServer.commit();
} catch (Exception e) {
e.printStackTrace();
}
}

}