應用activeMQ消息中間件同步索引庫

mq是一個消息服務器；

安裝包內置了tomcat，直接登錄訪問，登錄：http://ip:8161/admin/    （相當於dubbo的moniter監控中心） admin admin
傳統串行化，並行化：

mq消息服務器集中管理消息：

1）點對點：

 　　異步接收消息（監聽模式）：

 

2）訂閱模式：

　　必須使用監聽模式（異步）；

---

Spring整合mq：

　　配置生產者：

    <!-- 創建acitiveMQ消息服務工廠對象,把acitiveMQ消息服務器交給spring管理 -->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <constructor-arg name="brokerURL" value="tcp://192.168.74.132:61616"></constructor-arg>
    </bean>

    <!-- spring jms java消息服務提供工廠對象管理mq消息 -->
    <bean id="connectionFactory"
        class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="targetConnectionFactory"></property>
    </bean>

    <!-- spring jms 提供jms消息模版對象 發送消息 -->
    <bean class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"></property>
    </bean>

    <!-- 點對點模式空間 -->
    <!-- <bean id="oneQueue" class="org.apache.activemq.command.ActiveMQQueue"> 
        <constructor-arg value="oneQueue"></constructor-arg> </bean> -->
    <!-- 發布訂閱模式空間 -->
    <!-- <bean id="oneTopic" class="org.apache.activemq.command.ActiveMQTopic"> 
        <constructor-arg value="oneTopic"></constructor-arg> </bean> -->
    <bean id="add_update_del_topic" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="add_update_del_topic"></constructor-arg>
    </bean>

 

 　　配置消費者：

<!-- 創建acitiveMQ消息服務工廠對象,把acitiveMQ消息服務器交給spring管理 -->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <constructor-arg name="brokerURL" value="tcp://192.168.74.132:61616"></constructor-arg>
    </bean>

    <!-- spring jms java消息服務提供工廠對象管理mq消息 -->
    <bean id="connectionFactory"
        class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="targetConnectionFactory"></property>
    </bean>

    <!-- 點對點模式空間 -->
    <!-- <bean id="oneQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="oneQueue"></constructor-arg>
    </bean> -->
    <!-- 發布訂閱模式空間 -->
    <!-- <bean id="oneTopic" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="oneTopic"></constructor-arg>
    </bean> -->
    <bean id="add_update_del_topic" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="add_update_del_topic"></constructor-arg>
    </bean>

    <!-- 創建接受消息監聽器,此監聽器是真實接受消息的監聽器 -->
    <bean id="myMessageListener" class="cn.e3.search.listener.IndexListener"></bean>

    <!-- spring jms 提供消息監聽容器接受消息. -->
    <!-- 自動觸發接受消息 -->
    <bean
        class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <!-- 指定消息服務器地址 -->
        <property name="connectionFactory" ref="connectionFactory"></property>
        <!-- 指定接受消息服務器空間 -->
        <property name="destination" ref="add_update_del_topic"></property>
        <!-- 指定消息接受者 -->
        <property name="messageListener" ref="myMessageListener"></property>
    </bean>

---

 

 業務分析，面向服務分布式架構SOA：

商品發生變動后，發送消息至mq：

jmsTemplate.send(activeMQTopic,new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage(""+itemId);
            }
        });

  xml中配置監聽器，在監聽器類中根據接收的id查詢變動后的商品，將商品索引文檔寫入索引庫；

  <!-- 創建接受消息監聽器,此監聽器是真實接受消息的監聽器 -->
    <bean id="myMessageListener" class="cn.e3.search.listener.IndexListener"></bean>

    public void onMessage(Message message) {

        try {
            // 初始化一個商品id
            Long itemId = null;
            if (message instanceof TextMessage) {
                // 接受消息
                TextMessage m = (TextMessage) message;
                // 獲取商品id
                itemId = Long.parseLong(m.getText());
                //根據商品id查詢數據庫新的數據
                SearchItem searchItem = searchItemMapper.findDatabaseToSolrIndexWithID(itemId);
                
                //把數據庫數據封裝到文檔對象
                SolrInputDocument doc = new SolrInputDocument();
                //封裝文檔域字段所對應值
                //封裝文檔域所對應數據庫查詢值
                doc.addField("id", searchItem.getId());
                
                //標題
                doc.addField("item_title", searchItem.getTitle());
                //買點
                doc.addField("item_sell_point", searchItem.getSell_point());
                //價格
                doc.addField("item_price", searchItem.getPrice());
                //圖片地址
                doc.addField("item_image", searchItem.getImage());
                //商品類別
                doc.addField("item_category_name", searchItem.getCategory_name());
                //商品描述
                doc.addField("item_desc", searchItem.getItem_desc());
                
                //使用solr服務對象把索引文檔對象寫入索引庫,實現索引庫同步
                solrServer.add(doc);
                //提交
                solrServer.commit();
            }
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }