摘要：消費(fèi)者，監(jiān)聽生產(chǎn)者往指定目的地發(fā)送消息后，接下來(lái)就是消費(fèi)者對(duì)指定目的地的消息進(jìn)行消費(fèi)了。它不會(huì)動(dòng)態(tài)的適應(yīng)運(yùn)行時(shí)需要和參與外部的事務(wù)管理。它很好的平衡了對(duì)提供者要求低先進(jìn)功能如事務(wù)參與和兼容環(huán)境。

## ActiveMQ Config
activemq.brokerURL=tcp://127.0.0.1:61616
activemq.userName=admin
activemq.password=password
activemq.pool.maxConnection=10
activemq.queue=mailqueue
activemq.topic=mailtopic

    org.springframework
    spring-jms
    4.3.7.RELEASE



    org.apache.activemq
    activemq-pool
    5.15.3

ConnectionFactory 是用于產(chǎn)生到 JMS 服務(wù)器的鏈接的，Spring 為我們提供了多個(gè) ConnectionFactory，有 SingleConnectionFactory 和 CachingConnectionFactory。

SingleConnectionFactory ：對(duì)于建立 JMS 服務(wù)器鏈接的請(qǐng)求會(huì)一直返回同一個(gè)鏈接，并且會(huì)忽略 Connection 的 close 方法調(diào)用。

CachingConnectionFactory ：繼承了 SingleConnectionFactory，所以它擁有 SingleConnectionFactory 的所有功能，同時(shí)它還新增了緩存功能，它可以緩存 Session、MessageProducer 和 MessageConsumer。這里我們使用 CachingConnectionFactory 來(lái)作為示例。

import com.alibaba.fastjson.JSONObject;
import com.github.binarylei.jms.spring.core.Mail;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

/**
 * @author: leigang
 * @version: 2018-04-03
 */
@Service("mqProducer")
public class MQProducer {

    @Autowired
    private JmsTemplate jmsTemplate;

    public void sendMail(Mail mail) {
        jmsTemplate.send((session) -> {
            return session.createTextMessage(JSONObject.toJSONString(mail));
        });
    }
}

import com.alibaba.fastjson.JSONObject;
import com.github.binarylei.jms.spring.core.Mail;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

import javax.jms.Destination;

/**
 * @author: leigang
 * @version: 2018-04-03
 */
@Service("mqCustumer")
public class MQCustumer {

    @Autowired
    private JmsTemplate jmsTemplate;

    @Autowired
    private Destination destination;

    public void sendMail() {
        String json = (String) jmsTemplate.receiveAndConvert(destination);
        Mail mail = (Mail) JSONObject.parseObject(json, Mail.class);
        System.out.println(mail);
    }
}

我們?cè)?spring 中直接配置異步接收消息的監(jiān)聽器，這樣就相當(dāng)于在 spring 中配置了消費(fèi)者，在接受消息的時(shí)候就不必要啟動(dòng)消費(fèi)者了。

spring-jms.xml:

生產(chǎn)者往指定目的地 Destination 發(fā)送消息后，接下來(lái)就是消費(fèi)者對(duì)指定目的地的消息進(jìn)行消費(fèi)了。那么消費(fèi)者是如何知道有生產(chǎn)者發(fā)送消息到指定目的地 Destination了呢？這是通過(guò) Spring 為我們封裝的消息監(jiān)聽容器 MessageListenerContainer 實(shí)現(xiàn)的，它負(fù)責(zé)接收信息，并把接收到的信息分發(fā)給真正的 MessageListener 進(jìn)行處理。

每個(gè)消費(fèi)者對(duì)應(yīng)每個(gè)目的地都需要有對(duì)應(yīng)的 MessageListenerContainer。對(duì)于消息監(jiān)聽容器而言，除了要知道監(jiān)聽哪個(gè)目的地之外，還需要知道到哪里去監(jiān)聽，也就是說(shuō)它還需要知道去監(jiān)聽哪個(gè) JMS 服務(wù)器，這是通過(guò)在配置 MessageConnectionFactory 的時(shí)候往里面注入一個(gè) ConnectionFactory 來(lái)實(shí)現(xiàn)的。所以我們?cè)谂渲靡粋€(gè) MessageListenerContainer 的時(shí)候有三個(gè)屬性必須指定，一個(gè)是表示從哪里監(jiān)聽的 ConnectionFactory；一個(gè)是表示監(jiān)聽什么的 Destination；一個(gè)是接收到消息以后進(jìn)行消息處理的 MessageListener。

Spring 一共為我們提供了兩種類型的 MessageListenerContainer：

SimpleMessageListenerContainer ：SimpleMessageListenerContainer 會(huì)在一開始的時(shí)候就創(chuàng)建一個(gè)會(huì)話 session 和消費(fèi)者 Consumer，并且會(huì)使用標(biāo)準(zhǔn)的 JMS MessageConsumer.setMessageListener() 方法注冊(cè)監(jiān)聽器讓 JMS 提供者調(diào)用監(jiān)聽器的回調(diào)函數(shù)。它不會(huì)動(dòng)態(tài)的適應(yīng)運(yùn)行時(shí)需要和參與外部的事務(wù)管理。兼容性方面，它非常接近于獨(dú)立的 JMS 規(guī)范，但一般不兼容 Java EE 的 JMS 限制。

DefaultMessageListenerContainer ：在大多數(shù)情況下我們還是使用的 DefaultMessageListenerContainer，跟SimpleMessageListenerContainer 相比，DefaultMessageListenerContainer 會(huì)動(dòng)態(tài)的適應(yīng)運(yùn)行時(shí)需要，并且能夠參與外部的事務(wù)管理。它很好的平衡了對(duì) JMS 提供者要求低、先進(jìn)功能如事務(wù)參與和兼容 Java EE 環(huán)境。

消息監(jiān)聽器：

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.TextMessage;

/**
 * @author: leigang
 * @version: 2018-04-03
 */
public class MessageListener implements javax.jms.MessageListener  {

    @Override
    public void onMessage(Message message) {
        TextMessage msg = (TextMessage) message;
        try {
            System.out.println("消息：" + msg.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

在 spring-jms.xml 中將 ActiveMQTopic 生成 Topic，其它沒什么變化：

每天用心記錄一點(diǎn)點(diǎn)。內(nèi)容也許不重要，但習(xí)慣很重要！