這篇文章主要介紹了ActiveMQ訊息佇列技術融合Spring過程解析,文中通過示例程式碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

一、業務邏輯

我想在修改一個物品的狀態時，同時傳送廣播，給對應的監聽器去實現，此商品儲存到solr中，同時通過網頁靜態模板生成一個當前物品的詳情頁面，此時用到了廣播機制

當我刪除一個商品時，傳送一個廣播，給對應的監聽器，同時刪除solr中對應的物品。

廣播機制：必須要同時線上，才能接收我的訊息

使用訊息中介軟體需要匯入配置檔案

<?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.xsd
    http://www.springframework.org/schema/context  
    http://www.springframework.org/schema/context/spring-context.xsd">
  <!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供--> 
  <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> 
    <property name="brokerURL" value="tcp://192.168.200.128:61616"/> 
  </bean>
  <!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory --> 
  <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"> 
  <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory --> 
    <property name="targetConnectionFactory" ref="targetConnectionFactory"/> 
  </bean>
  <!-- Spring提供的JMS工具類，它可以進行訊息傳送、接收等 --> 
  <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> 
    <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory物件 --> 
    <property name="connectionFactory" ref="connectionFactory"/> 
  </bean>  
  
  <!-- 釋出訂閱模式,商品匯入索引庫和生成靜態頁面 -->
  <bean id="topicPageAndSolrDestination" class="org.apache.activemq.command.ActiveMQTopic">
     <!--將商品上架所有的商品的id傳送到這個佇列中-->
     <constructor-arg value="youlexuan_topic_page_solr"/>
  </bean>
  <!-- 點對點模式-->
  <bean id="queueSolrDeleteDestination" class="org.apache.activemq.command.ActiveMQQueue">
    <!--將商品上架所有的商品的id傳送到這個佇列中-->
    <constructor-arg value="youlexuan_queue_solr_delete"/>
  </bean> 
  
</beans>

釋出廣播：

if ("1".equals(status)){
  jmsTemplate.send(topicPageAndSolrDestination,new MessageCreator() {
    @Override
    public Message createMessage(Session session) throws JMSException {
      TextMessage textMessage = session.createTextMessage(String.valueOf(id));
      return textMessage;
    }
  });
}

監聽器1，將當前商品存入solr中:操作solr的伺服器配置檔案

<?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.xsd
    http://www.springframework.org/schema/context  
    http://www.springframework.org/schema/context/spring-context.xsd">
   <!--產生Connection-->
  <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> 
    <property name="brokerURL" value="tcp://192.168.200.128:61616"/> 
  </bean>
  <!--spring 管理connectionFactory-->
  <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
    <property name="targetConnectionFactory" ref="targetConnectionFactory"/> 
  </bean>
  <!--釋出訂閱模式  將資料匯入solr 索引庫-->
  <bean id="topicPageAndSolrDestination" class="org.apache.activemq.command.ActiveMQTopic"> 
    <constructor-arg value="youlexuan_topic_page_solr"/>
  </bean>
  <!--釋出訂閱模式  訊息監聽容器 將資料匯入solr 索引庫-->
  <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="destination" ref="topicPageAndSolrDestination" />
    <property name="messageListener" ref="pageAndSolrListener" />
  </bean>
#對應的用來監聽執行往solr中儲存庫存的訊息
  <bean id="pageAndSolrListener" class="com.ghh.sellergoods.service.listener.ItemSearchListener"></bean>
  <!--點對點的模式 刪除索引庫-->
  <bean id="queueSolrDeleteDestination" class="org.apache.activemq.command.ActiveMQQueue">
    <!--指定從這個佇列中 接收下架商品的-->
    <constructor-arg value="youlexuan_queue_solr_delete"/>
  </bean>
  <!--點對點的模式 訊息監聽器 刪除索引庫-->
  <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="destination" ref="queueSolrDeleteDestination" />
    <property name="messageListener" ref="itemDeleteListener" />
  </bean>
  <bean id="itemDeleteListener" class="com.ghh.sellergoods.service.listener.ItemDeleteListener"></bean>
</beans>

監聽器類

public class ItemSearchListener implements MessageListener {
  @Autowired
  private SearchService searchService;
  @Autowired
  private ItemDao itemDao;
  @Override
  public void onMessage(Message message) {
    //獲取生產者釋出的廣播，往solr中新增庫存列表
    ActiveMQTextMessage atm = (ActiveMQTextMessage) message;
    try {
      //獲取廣播中的資料。
      Long goodsId = Long.valueOf(atm.getText());
      //通過傳過來的商品id去查詢庫存表
      ItemQuery query = new ItemQuery();
      ItemQuery.Criteria criteria = query.createCriteria();
      criteria.andGoodsIdEqualTo(goodsId);
      //查詢對應商品id的庫存表
      List<Item> items = itemDao.selectByExample(query);
　　　　　　　　//呼叫對應的方法，往solr中添加當前商品對應庫存資訊
      searchService.importList(items);
    } catch (JMSException e) {
      e.printStackTrace();
    }
  }
}

監聽器類2：配置檔案

<?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.xsd
    http://www.springframework.org/schema/context  
    http://www.springframework.org/schema/context/spring-context.xsd"> 
  <!--產生Connection工廠類-->
  <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://192.168.200.128:61616"/> 
  </bean>
  <!--spring管理工廠類-->
  <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
    <property name="targetConnectionFactory" ref="targetConnectionFactory"/> 
  </bean>
  <!--釋出訂閱模式 生成頁面-->
  <bean id="topicPageAndSolrDestination" class="org.apache.activemq.command.ActiveMQTopic"> 
     <!--指定從這個佇列上獲取上架的商品id-->
     <constructor-arg value="youlexuan_topic_page_solr"/>
  </bean>
  <!--釋出訂閱模式 訊息監聽器 生成頁面-->
  <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="destination" ref="topicPageAndSolrDestination" />
    <property name="messageListener" ref="pageListener" />
  </bean>
  <bean id="pageListener" class="com.ghh.core.service.listener.PageListener"></bean>
  
</beans>

監聽器類2：生成靜態網頁模板

public class PageListener implements MessageListener {
  @Autowired
  private CmsService cmsService;
  @Override
  public void onMessage(Message message) {
    ActiveMQTextMessage atm = (ActiveMQTextMessage) message;
    try {
      Long goodsId = Long.valueOf(atm.getText());
      Map<String,Object> goodsData = cmsService.findGoodsData(goodsId);
      cmsService.createStaticPage(goodsId,goodsData);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
}

點對點

當我刪除商品時，我需要對應的服務進行刪除solr中庫存資訊，新增和刪除使用的是同一個服務中，使用的是上面的配置檔案

//釋出廣播,@Autowired
  private ActiveMQTopic topicPageAndSolrDestination;
//在修改的程式碼方法中來廣播發布當前商品的id

if (ids.length>0) {
      jmsTemplate.send(queueSolrDeleteDestination,new MessageCreator() {
        @Override
        public Message createMessage(Session session) throws JMSException {
          TextMessage textMessage = session.createTextMessage(String.valueOf(ids));
          return textMessage;
        }
      });
    }

#執行刪除solr中庫存資訊

public class ItemDeleteListener implements MessageListener {
  @Autowired
  private SearchService searchService;

  @Override
  public void onMessage(Message message) {
    ActiveMQTextMessage atm = (ActiveMQTextMessage) message;
    try {
      Long goodsId = Long.valueOf(atm.getText());
      searchService.deleteById(goodsId);
    } catch (JMSException e) {
      e.printStackTrace();
    }
  }
}

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支援我們。