集成JMS

    所谓消息服务，就是两个进程之间，通过消息服务器传递消息：

    使用消息服务，而不是直接调用对方的API，它的好处是：

    • 双方各自无需知晓对方的存在，消息可以异步处理，因为消息服务器会在Consumer离线的时候自动缓存消息；
    • 如果Producer发送的消息频率高于Consumer的处理能力，消息可以积压在消息服务器，不至于压垮Consumer；
    • 通过一个消息服务器，可以连接多个Producer和多个Consumer。

    因为消息服务在各类应用程序中非常有用，所以JavaEE专门定义了JMS规范。注意到JMS是一组接口定义，如果我们要使用JMS，还需要选择一个具体的JMS产品。常用的JMS服务器有开源的，商业服务器如WebLogic、WebSphere等也内置了JMS支持。这里我们选择开源的ActiveMQ作为JMS服务器，因此，在开发JMS之前我们必须首先安装ActiveMQ。

    现在问题来了：从官网下载ActiveMQ时，蹦出一个页面，让我们选择ActiveMQ Classic或者ActiveMQ Artemis，这两个是什么关系，又有什么区别？

    实际上ActiveMQ Classic原来就叫ActiveMQ，是Apache开发的基于JMS 1.1的消息服务器，目前稳定版本号是5.x，而ActiveMQ Artemis是由RedHat捐赠的HornetQ服务器代码的基础上开发的，目前稳定版本号是2.x。和ActiveMQ Classic相比，Artemis版的代码与Classic完全不同，并且，它支持JMS 2.0，使用基于Netty的异步IO，大大提升了性能。此外，Artemis不仅提供了JMS接口，它还提供了AMQP接口，STOMP接口和物联网使用的MQTT接口。选择Artemis，相当于一鱼四吃。

    所以，我们这里直接选择ActiveMQ Artemis。从官网最新的2.x版本，解压后设置环境变量，指向Artemis根目录，例如C:\Apps\artemis，然后，把ARTEMIS_HOME/bin加入PATH环境变量：

    • Windows下添加%ARTEMIS_HOME%\bin到Path路径；
    • Mac和Linux下添加$ARTEMIS_HOME/bin到PATH路径。

    Artemis有个很好的设计，就是它把程序和数据完全分离了。我们解压后的ARTEMIS_HOME目录是程序目录，要启动一个Artemis服务，还需要创建一个数据目录。我们把数据目录直接设定在项目spring-integration-jms的jms-data目录下。执行命令artemis create jms-data：

    1. $ pwd
    2. /Users/liaoxuefeng/workspace/spring-integration-jms
    4. $ artemis create jms-data
    5. Creating ActiveMQ Artemis instance at: /Users/liaoxuefeng/workspace/spring-integration-jms/jms-data
    7. --user: is a mandatory property!
    8. Please provide the default username:
    9. admin
    11. --password: is mandatory with this configuration:
    12. Please provide the default password:
    13. ********
    15. --allow-anonymous | --require-login: is a mandatory property!
    16. Allow anonymous access?, valid values are Y,N,True,False
    17. N
    19. Auto tuning journal ...
    20. done! Your system can make 0.09 writes per millisecond, your journal-buffer-timeout will be 11392000
    22. You can now start the broker by executing:  
    24.    "/Users/liaoxuefeng/workspace/spring-integration-jms/jms-data/bin/artemis" run
    26. Or you can run the broker in the background using:
    28.    "/Users/liaoxuefeng/workspace/spring-integration-jms/jms-data/bin/artemis-service" start

    在创建过程中，会要求输入连接用户和口令，这里我们设定admin和password，以及是否允许匿名访问（这里选择N）。

    此数据目录jms-data不仅包含消息数据、日志，还自动创建了两个启动服务的命令bin/artemis和bin/artemis-service，前者在前台启动运行，按Ctrl+C结束，后者会一直在后台运行。

    我们把目录切换到jms-data/bin，直接运行artemis run即可启动Artemis服务：

    1. $ ./artemis run
    2.      _        _               _
    3.     / \  ____| |_  ___ __  __(_) _____
    4.    / _ \|  _ \ __|/ _ \  \/  | |/  __/
    5.   / ___ \ | \/ |_/  __/ |\/| | |\___ \
    6.  /_/   \_\|   \__\____|_|  |_|_|/___ /
    7.  Apache ActiveMQ Artemis 2.13.0
    9. ...
    11. 2020-06-02 07:50:21,718 INFO  [org.apache.activemq.artemis] AMQ241001: HTTP Server started at http://localhost:8161
    12. 2020-06-02 07:50:21,718 INFO  [org.apache.activemq.artemis] AMQ241002: Artemis Jolokia REST API available at http://localhost:8161/console/jolokia
    13. 2020-06-02 07:50:21,719 INFO  [org.apache.activemq.artemis] AMQ241004: Artemis Console available at http://localhost:8161/console

    启动成功后，Artemis提示可以通过URLhttp://localhost:8161/console访问管理后台。注意不要关闭命令行窗口。

    在编写JMS代码之前，我们首先得理解JMS的消息模型。JMS把生产消息的一方称为Producer，处理消息的一方称为Consumer。有两种类型的消息通道，一种是Queue：

    1. │Producer│───>│ Queue  │───>│Consumer│
    2. └────────┘    └────────┘    └────────┘

    一种是Topic：

    1.                             ┌────────┐
    2.                          ┌─>│Consumer│
    3.                          │  └────────┘
    5. │Producer│───>│ Topic  │─┼─>│Consumer│
    6. └────────┘    └────────┘ │  └────────┘
    7.                          │  ┌────────┐
    8.                          └─>│Consumer│
    9.                             └────────┘

    它们的区别在于，Queue是一种一对一的通道，如果Consumer离线无法处理消息时，Queue会把消息存起来，等Consumer再次连接的时候发给它。设定了持久化机制的Queue不会丢失消息。如果有多个Consumer接入同一个Queue，那么它们等效于以集群方式处理消息，例如，发送方发送的消息是A，B，C，D，E，F，两个Consumer可能分别收到A，C，E和B，D，F，即每个消息只会交给其中一个Consumer处理。

    Topic则是一种一对多通道。一个Producer发出的消息，会被多个Consumer同时收到，即每个Consumer都会收到一份完整的消息流。那么问题来了：如果某个Consumer暂时离线，过一段时间后又上线了，那么在它离线期间产生的消息还能不能收到呢？

    这取决于消息服务器对Topic类型消息的持久化机制。如果消息服务器不存储Topic消息，那么离线的Consumer会丢失部分离线时期的消息，如果消息服务器存储了Topic消息，那么离线的Consumer可以收到自上次离线时刻开始后产生的所有消息。JMS规范通过Consumer指定一个持久化订阅可以在上线后收取所有离线期间的消息，如果指定的是非持久化订阅，那么离线期间的消息会全部丢失。

    细心的童鞋可以看出来，如果一个Topic的消息全部都持久化了，并且只有一个Consumer，那么它和Queue其实是一样的。实际上，很多消息服务器内部都只有Topic类型的消息架构，Queue可以通过Topic“模拟”出来。

    以上是消息服务的基本模型。具体到某个消息服务器时，Producer和Consumer通常是通过TCP连接消息服务器，在编写JMS程序时，又会遇到ConnectionFactory、Connection、Session等概念，其实这和JDBC连接是类似的：

    • ConnectionFactory：代表一个到消息服务器的连接池，类似JDBC的DataSource；
    • Connection：代表一个到消息服务器的连接，类似JDBC的Connection；
    • Session：代表一个经过认证后的连接会话；
    • Message：代表一个消息对象。

    在JMS 1.1中，发送消息的典型代码如下：

    JMS 2.0改进了一些API接口，发送消息变得更简单：

    1. try (JMSContext context = connectionFactory.createContext()) {
    2.     context.createProducer().send(queue, text);
    3. }

    JMSContext实现了AutoCloseable接口，可以使用try(resource)语法，代码更简单。

    有了以上预备知识，我们就可以开始开发JMS应用了。

    首先，我们在pom.xml中添加如下依赖：

    • org.springframework:spring-jms:5.2.0.RELEASE
    • javax.jms:javax.jms-api:2.0.1
    • org.apache.activemq:artemis-jms-client:2.13.0
    • io.netty:netty-handler-proxy:4.1.45.Final

    在AppConfig中，通过@EnableJms让Spring自动扫描JMS相关的Bean，并加载JMS配置文件jms.properties：

    1. @Configuration
    2. @ComponentScan
    3. @EnableWebMvc
    4. @EnableJms // 启用JMS
    5. @EnableTransactionManagement
    6. @PropertySource({ "classpath:/jdbc.properties", "classpath:/jms.properties" })
    7. public class AppConfig {
    8.     ...
    9. }

    首先要创建的Bean是ConnectionFactory，即连接消息服务器的连接池：

    1. @Bean
    2. ConnectionFactory createJMSConnectionFactory(
    3.     @Value("${jms.uri:tcp://localhost:61616}") String uri,
    4.     @Value("${jms.username:admin}") String username,
    5.     @Value("${jms.password:password}") String password)
    6. {
    7.     return new ActiveMQJMSConnectionFactory(uri, username, password);
    8. }

    因为我们使用的消息服务器是ActiveMQ Artemis，所以ConnectionFactory的实现类就是消息服务器提供的ActiveMQJMSConnectionFactory，它需要的参数均由配置文件读取后传入，并设置了默认值。

    我们再创建一个JmsTemplate，它是Spring提供的一个工具类，和JdbcTemplate类似，可以简化发送消息的代码：

    1. @Bean
    2. JmsTemplate createJmsTemplate(@Autowired ConnectionFactory connectionFactory) {
    3.     return new JmsTemplate(connectionFactory);
    4. }

    下一步要创建的是JmsListenerContainerFactory，

    除了必须指定Bean的名称为jmsListenerContainerFactory外，这个Bean的作用是处理和Consumer相关的Bean。我们先跳过它的原理，继续编写MessagingService来发送消息：

    1. @Component
    2. public class MessagingService {
    3.     @Autowired ObjectMapper objectMapper;
    4.     @Autowired JmsTemplate jmsTemplate;
    6.     public void sendMailMessage(MailMessage msg) throws Exception {
    7.         jmsTemplate.send("jms/queue/mail", new MessageCreator() {
    8.             public Message createMessage(Session session) throws JMSException {
    9.                 return session.createTextMessage(text);
    10.             }
    11.         });
    12.     }
    13. }

    JMS的消息类型支持以下几种：

    • TextMessage：文本消息；
    • BytesMessage：二进制消息；
    • MapMessage：包含多个Key-Value对的消息；
    • ObjectMessage：直接序列化Java对象的消息；
    • StreamMessage：一个包含基本类型序列的消息。

    最常用的是发送基于JSON的文本消息，上述代码通过JmsTemplate创建一个TextMessage并发送到名称为jms/queue/mail的Queue。

    注意：Artemis消息服务器默认配置下会自动创建Queue，因此不必手动创建一个名为jms/queue/mail的Queue，但不是所有的消息服务器都会自动创建Queue，生产环境的消息服务器通常会关闭自动创建功能，需要手动创建Queue。

    再注意到MailMessage是我们自己定义的一个JavaBean，真正的JMS消息是创建的，它的内容是JSON。

    当用户注册成功后，我们就调用MessagingService.sendMailMessage()发送一条JMS消息，此代码十分简单，这里不再贴出。

    处理消息的核心代码是编写一个Bean，并在处理方法上标注@JmsListener：

    1. @Component
    2. public class MailMessageListener {
    3.     final Logger logger = LoggerFactory.getLogger(getClass());
    5.     @Autowired ObjectMapper objectMapper;
    6.     @Autowired MailService mailService;
    8.     @JmsListener(destination = "jms/queue/mail", concurrency = "10")
    9.     public void onMailMessageReceived(Message message) throws Exception {
    10.         logger.info("received message: " + message);
    11.         if (message instanceof TextMessage) {
    12.             String text = ((TextMessage) message).getText();
    13.             MailMessage mm = objectMapper.readValue(text, MailMessage.class);
    14.             mailService.sendRegistrationMail(mm);
    15.         } else {
    16.             logger.error("unable to process non-text message!");
    17.         }
    18.     }
    19. }

    注意到@JmsListener指定了Queue的名称，因此，凡是发到此Queue的消息都会被这个onMailMessageReceived()方法处理，方法参数是JMS的Message接口，我们通过强制转型为TextMessage并提取JSON，反序列化后获得自定义的JavaBean，也就获得了发送邮件所需的所有信息。

    下面问题来了：Spring处理JMS消息的流程是什么？

    如果我们直接调用JMS的API来处理消息，那么编写的代码大致如下：

    1. // 创建JMS连接:
    2. Connection connection = connectionFactory.createConnection();
    3. // 创建会话:
    4. Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    5. // 创建一个Consumer:
    6. MessageConsumer consumer = session.createConsumer(queue);
    7. // 为Consumer指定一个消息处理器:
    8. consumer.setMessageListener(new MessageListener() { 
    9.     public void onMessage(Message message) {
    10.         // 在此处理消息... 
    11.     }
    12. });
    13. // 启动接收消息的循环:
    14. connection.start();

    我们自己编写的MailMessageListener.onMailMessageReceived()相当于消息处理器：

    1. consumer.setMessageListener(new MessageListener() { 
    2.     public void onMessage(Message message) {
    3.         mailMessageListener.onMailMessageReceived(message); 
    4.     }

    所以，Spring根据AppConfig的注解@EnableJms自动扫描带有@JmsListener的Bean方法，并为其创建一个MessageListener把它包装起来。

    注意到前面我们还创建了一个JmsListenerContainerFactory的Bean，它的作用就是为每个MessageListener创建MessageConsumer并启动消息接收循环。

    再注意到@JmsListener还有一个concurrency参数，10表示可以最多同时并发处理10个消息，5-10表示并发处理的线程可以在5~10之间调整。

    因此，Spring在通过MessageListener接收到消息后，并不是直接调用mailMessageListener.onMailMessageReceived()，而是用线程池调用，因此，要时刻牢记，onMailMessageReceived()方法可能被多线程并发执行，一定要保证线程安全。

    我们总结一下Spring接收消息的步骤：

    通过JmsListenerContainerFactory配合@EnableJms扫描所有@JmsListener方法，自动创建MessageConsumer、MessageListener以及线程池，启动消息循环接收处理消息，最终由我们自己编写的@JmsListener方法处理消息，可能会由多线程同时并发处理。

    要验证消息发送和处理，我们注册一个新用户，可以看到如下日志输出：

    可见，消息被成功发送到Artemis，然后在很短的时间内被接收处理了。

    使用消息服务对发送Email进行改造的好处是，发送Email的能力通常是有限的，通过JMS消息服务，如果短时间内需要给大量用户发送Email，可以先把消息堆积在JMS服务器上慢慢发送，对于批量发送邮件、短信等尤其有用。

    从下载练习： （推荐使用IDE练习插件快速下载）

    JMS是Java消息服务，可以通过JMS服务器实现消息的异步处理。

    消息服务主要解决Producer和Consumer生产和处理速度不匹配的问题。