activemq和spring的整合应用

我会带着你远行 2022-05-22 01:13 164阅读 0赞

应用场景：当我们给商品添加一条记录后，需要把商品的信息同步到索引库。

##### 1：配置producer，用*jmsTemplate去发送消息（queueDestination这个名称需要和监听中配置对应）* #####

<beanid=*"targetConnectionFactory"*class=*"org.apache.activemq.ActiveMQConnectionFactory"*>

</bean>

<beanid=*"connectionFactory"*

class=*"org.springframework.jms.connection.SingleConnectionFactory"*>

</bean>

</bean>

<constructor-arg>

<value>spring-queue</value>

</constructor-arg>

</bean>

<constructor-arg value=*"topic"* />

</bean>

</beans>

#####  #####

##### 2：当商品添加完成后发送一个TextMessage，包含一个商品id。，模板和发送目的地用spring注入的方式 #####

@Autowired

private JmsTemplate  jmsTemplate;

@Autowired

private Destination topicDestination;

<table> 
 <tbody> 
 <tr> 
 <td>@Override &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; public TaotaoResult addItem(TbItem item, String desc) { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // 1、生成商品id &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; finallongitemId = IDUtils.genItemId(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // 2、补全TbItem对象的属性 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; item.setId(itemId); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //商品状态，1-正常，2-下架，3-删除 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; item.setStatus((byte) 1); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; Date date = new Date(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; item.setCreated(date); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; item.setUpdated(date); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // 3、向商品表插入数据 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; itemMapper.insert(item); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // 4、创建一个TbItemDesc对象 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; TbItemDesc itemDesc = new TbItemDesc(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // 5、补全TbItemDesc的属性 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; itemDesc.setItemId(itemId); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; itemDesc.setItemDesc(desc); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; itemDesc.setCreated(date); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; itemDesc.setUpdated(date); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // 6、向商品描述表插入数据 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; itemDescMapper.insert(itemDesc); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; //发送一个商品添加消息 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; jmsTemplate.send(topicDestination, new MessageCreator() { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; @Override &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; public Message createMessage(Session session) throws JMSException { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; TextMessage textMessage = session.createTextMessage(itemId + ""); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; returntextMessage; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; } &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // 7、TaotaoResult.ok() &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return TaotaoResult.ok(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }</td> 
 </tr> 
 </tbody> 
</table>

##### 3:写监听，配置监听 #####

**publicclass**ItemChangeListener **implements** MessageListener \{

@Autowired

**private**SearchItemServiceImpl searchItemServiceImpl;

@Override

**publicvoid** onMessage(Message message) \{

**try** \{

TextMessage textMessage = **null**;

Long itemId = **null**;

//取商品id

**if** (message**instanceof** TextMessage) \{

textMessage = (TextMessage) message;

itemId = Long.*parseLong*(textMessage.getText());

//向索引库添加文档

searchItemServiceImpl.addDocument(itemId);

\} **catch** (Exception e) \{

e.printStackTrace();

<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> 
<property name="brokerURL" value="tcp://192.168.25.168:61616" /> 
</bean> 
 
<bean id="connectionFactory" 
class="org.springframework.jms.connection.SingleConnectionFactory"> 
 
<property name="targetConnectionFactory" ref="targetConnectionFactory" /> 
</bean> 
 
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic"> 
<constructor-arg value="item-add-topic" /> 
</bean> 
 
 
<bean id="htmlGenListener" class="com.taotao.item.listener.HtmlGenListener"/> 
 
 
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer"> 
<property name="connectionFactory" ref="connectionFactory" /> 
<property name="destination" ref="topicDestination" /> 
<property name="messageListener" ref="htmlGenListener" /> 
</bean> 
</beans>

##### 3；serivce层 #####

**public** TaotaoResult addDocument(**long**itemId) **throws** Exception \{

// 1、根据商品id查询商品信息。

SearchItem searchItem = searchItemMapper.getItemById(itemId);

// 2、创建一SolrInputDocument对象。

SolrInputDocument document = **new** SolrInputDocument();

// 3、使用SolrServer对象写入索引库。

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());