Spring源码——消息队列

前言

内容主要参考自《Spring源码深度解析》一书，算是读书笔记或是原书的补充。进入正文后可能会引来各种不适，毕竟阅读源码是件极其痛苦的事情。

本文主要涉及书中第十三章的部分，依照书中内容以及个人理解对Spring源码进行了注释，详见Github仓库：https://github.com/MrSorrow/spring-framework

Java消息服务 (Java Message Service，JMS) 应用程序接口是一个Java平台中关于面向消息中间件(MOM)的API，用于在两个应用程序之间或分布式系统中发送消息进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数 MOM 提供商都对 JMS 提供支持。

Java消息服务的规范包括两种消息模式，点对点和发布者/订阅者。许多提供商支持这一通用框架。因此，程序员可以在他们的分布式软件中实现面向消息的操作，这些操作将具有不同面向消息中间件产品的可移植性。

Java消息服务支持同步和异步的消息处理，在某些场景下，异步消息是必要的，而且比同步消息操作更加便利。

本文以 Java 消息服务的开源实现产品 ActiveMQ 为例来进行Spring整合消息服务功能的实现分析。

I. 单独使用ActiveMQ

安装ActiveMQ

这里我是在Windows平台上安装 ActiveMQ 的。需要等上官网，下载 apache-activemq-5.15.7-bin.zip 。下载完毕后解压至本地磁盘，运行 apache-activemq-5.15.7\bin\win64\activemq.bat 批处理文件，ActiveMQ 就能够顺利启动了。

Spring源码——消息队列

访问 http://localhost:8161/admin，登录默认账号密码都为 admin。

Spring源码——消息队列

JMS独立使用

尽管大多数的Java消息服务的使用都会跟Spring相结合，但是，我们还是非常有必要了解消息的独立使用方法，这对于我们了解消息的实现原理以及后续的与Spring整合实现分析都非常重要。消息服务的使用除了要开启消息服务器外，还需要构建消息的发送端与接收端，发送端主要用来将包含业务逻辑的消息发送至消息服务器，而消息接收端则用于将服务器中的消息提取并进行相应的处理。

① 发送端

发送端主要用于发送消息到消息服务器，以下为发送消息测试，尝试发送 10 条消息到消息服务器，消息的内容为“测试发送消息”。

public class Sender {

   public static void main(String[] args) throws JMSException, InterruptedException {
      ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();

      Connection connection = connectionFactory.createConnection();
      Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
      Destination destination = session.createQueue("my-queue");

      MessageProducer producer = session.createProducer(destination);
      for (int i = 0; i < 10; i++) {
         TextMessage message = session.createTextMessage("测试发送消息");
         Thread.sleep(1000);
         producer.send(message);
      }

      session.commit();
      session.close();
      connection.close();
   }
}

② 接收端

接收端主要用于连接消息服务器并接收服务器上的消息。

public class Receiver {

   public static void main(String[] args) throws JMSException {
      ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();

      Connection connection = connectionFactory.createConnection();
      connection.start();

      Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
      Destination destination = session.createQueue("my-queue");

      MessageConsumer consumer = session.createConsumer(destination);

      int i = 0;
      while (i < 10) {
         i++;
         TextMessage message = (TextMessage) consumer.receive();
         session.commit();
         System.out.println("接收到消息内容为：" + message.getText());
      }

      session.close();
      connection.close();
   }
}

③ 测试结果

先运行发送端向消息队列中发送 10 条消息，然后运行接收端，即可打印出发送端向 my-queue 发送的 10 条消息内容。接收端总共消费了10次消息，消息队列中 my-queue 中的消息应该全部被消费完毕。

分析

从发送端与接收端的代码可以看出，整个消息的发送与接收过程非常简单，但是其中却参杂着大量的冗余代码，比如 Connection 的创建与关闭，Session 的创建与关闭等。

对于发送的流程，主要包括：

初始化连接工厂；
利用连接工厂创建一个连接；
使用连接建立会话 Session；
通过会话创建一个管理对象 Destination ，包括队列 (Queue) 或主题 (Topic)；
使用会话 Session 和管理对象 Destination 创建消息生产者 MessageSender；
使用消息生产者 MessageSender 发送消息。

对于接收的流程，主要包括：

1-4 步与发送相同；
使用会话 Session 和管理对象 Destination 创建消息消费者 MessageConsumer；
使用消息消费者 MessageConsumer 接收消息。

很容易让我们联想到数据库JDBC的实现，在使用消息队列时都需要一系列冗余的但又必不可少的套路代码，而其中真正用于数据操作/发送消息的代码其实很简单。前 1-3 步都可以Spring帮助我们完成，包含个性化信息的步骤交给用户进行设置。

所以Spring对于 JMS 消息队列同样利用模板设计模式加上回调的方式提供了一个模板类 JmsTemplate，能让我们非常快捷方便地利用Spring进行消息的收发。

II. Spring整合ActiveMQ

和Spring封装JDBC一样，Spring也提供了一个模板类 JmsTemplate 来帮助我们使用 JMS。

添加依赖

主要在环境中添加上 spring-jms 和 activemq 两个依赖即可。

plugins {
    id 'java'
}

group 'org.springframework'
version '5.1.0.BUILD-SNAPSHOT'

sourceCompatibility = 1.8

repositories {
    mavenCentral()
}

dependencies {
    compile(project(":spring-beans"))
    compile(project(":spring-context"))
    compile(project(":spring-aop"))
    compile(project(":spring-jdbc"))
    compile(project(":spring-jms"))
    compile group: 'org.springframework', name: 'spring-aspects', version: '5.0.7.RELEASE'
    compile group: 'org.apache.commons', name: 'commons-dbcp2', version: '2.5.0'
    compile group: 'mysql', name: 'mysql-connector-java', version: '5.1.18'
    compile group: 'org.mybatis', name: 'mybatis', version: '3.4.6'
    compile group: 'org.mybatis', name: 'mybatis-spring', version: '1.3.2'
    compile group: 'org.apache.activemq', name: 'activemq-pool', version: '5.15.7'
    testCompile group: 'junit', name: 'junit', version: '4.12'
}

sourceSets.main.resources.srcDirs = ["src/main/java","src/main/resources"]

配置文件

Spring整合消息服务的使用也从配置文件配置开始。在 Spring 的核心配置文件中首先要注册 JmsTemplate 类型的 bean。当然，ActiveMQConnectionFactory 用于连接消息服务器，是消息服务的基础，也要注册 ActiveMQQueue 则用于指定消息的目的地。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

   <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
      <property name="brokerURL">
         <value>tcp://127.0.0.1:61616</value>
      </property>
   </bean>

   <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
      <property name="connectionFactory">
         <ref bean="connectionFactory" />
      </property>
   </bean>

   <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
      <constructor-arg index="0">
         <value>spring_and_activemq</value>
      </constructor-arg>
   </bean>
</beans>

发送端

有了以上的配置，Spring就可以根据配置信息简化我们的工作量。Spring中使用发送消息到消息服务器，省去了冗余的 Connection 以及 Session 等的创建与销毁过程，简化了工作量。

public class SpringSender {

	@Test
	public void sendMessage() {
		ApplicationContext context = new ClassPathXmlApplicationContext("activeMQ-Test.xml");
		JmsTemplate jmsTemplate = (JmsTemplate) context.getBean("jmsTemplate");
		Destination destination = (Destination) context.getBean("destination");

		jmsTemplate.send(destination, new MessageCreator() {
			@Override
			public Message createMessage(Session session) throws JMSException {
				return session.createTextMessage("发送一个消息");
			}
		});
	}
}

接收端

同样，在Spring中接收消息也非常方便，Spring中连接服务器接收消息的示例如下。

public class SpringReceiver {

   @Test
   public void receiveMessage() {
      ApplicationContext context = new ClassPathXmlApplicationContext("activeMQ-Test.xml");
      JmsTemplate jmsTemplate = (JmsTemplate) context.getBean("jmsTemplate");
      Destination destination = (Destination) context.getBean("destination");

      TextMessage textMessage = (TextMessage) jmsTemplate.receive(destination);
      System.out.println("接收到消息：" + textMessage);
   }
}

测试结果

同样，先运行发送端程序，然后运行接收端代码，测试结果如下。

Spring源码——消息队列

监听器

使用 jmsTemplate.receive(destination) 方法只能接收一次消息，如果未接收到消息，则会一直等待，当然用户可以通过设置 timeout 属性来控制等待时间，但是一旦接收到消息本次接收任务就会结束，虽然用户可以通过 while(true) 的方式来实现循环监听消息服务器上的消息，还有一种更好的解决办法：创建消息监听器。消息监听器的使用方式如下。

① 创建消息监听器

用于监听消息，一旦有新消息Spring会将消息引导至消息监听器以方便用户进行相应的逻辑处理。实现监听器需要实现 MessageListener 接口，重写 onMessage() 方法。

public class MyMessageListener implements MessageListener {

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

② 修改配置文件

注入自定义的监听器 bean，添加一个监听器容器。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">


   <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
      <property name="brokerURL">
         <value>tcp://127.0.0.1:61616</value>
      </property>
   </bean>

   <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
      <property name="connectionFactory">
         <ref bean="connectionFactory" />
      </property>
   </bean>

   <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
      <constructor-arg index="0">
         <value>spring_and_activemq</value>
      </constructor-arg>
   </bean>

   <bean id="myMessageListener" class="guo.ping.activemq.MyMessageListener" />

   <bean id="javaConsumer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
      <property name="connectionFactory" ref="connectionFactory" />
      <property name="destination" ref="destination" />
      <property name="messageListener" ref="myMessageListener" />
   </bean>

</beans>

③ 测试结果

通过以上的修改便可以进行消息监听的功能了，一旦有消息传入至消息服务器，则会被消息监听器监听到，并由Spring将消息内容引导至消息监听器的处理函数中等待用户的进一步逻辑处理。

将发送代码改为循环发送10条消息，可以看到结果如下，只截取了部分，可以看到监听器一直在接收消息。

Spring源码——消息队列

消息接收可以使用消息监听器的方式替代模版方法，但是在发送消息的时候是无法替代的。接下来，我们主要研究 JmsTemplate 中的发送接收消息方法。

III. 源码分析

查看 JmsTemplate 的类型层级结构图，发现其实现了 InitializingBean 接口。

Spring源码——消息队列

实现 InitializingBean 接口的方法是在 JmsAccessor 抽象类中，实现内容如下。

/**
 * 实现InitializingBean接口
 */
@Override
public void afterPropertiesSet() {
   // 对ConnectionFactory判空
   if (getConnectionFactory() == null) {
      throw new IllegalArgumentException("Property 'connectionFactory' is required");
   }
}

/**
 * Return the ConnectionFactory that this accessor uses for obtaining
 * JMS {@link Connection Connections}.
 */
@Nullable
public ConnectionFactory getConnectionFactory() {
    return this.connectionFactory;
}

方法中只是一个验证连接工厂存在与否的功能，并没有其他逻辑实现。所以，创建 JmsTemplate 没有什么特殊的 bean 后处理等操作，我们可以直接进行 JmsTemplate 模板类中方法的分析。

JmsTemplate

我们先以发送方法为例，使用模板类的 send() 方法示例如下。

jmsTemplate.send(destination, new MessageCreator() {
   @Override
   public Message createMessage(Session session) throws JMSException {
      return session.createTextMessage("发送一个消息");
   }
});

可以看到方法中传入了两个参数，一个是 ActiveMQQueue 类型的 bean——destination，另一个是 MessageCreator 接口的实现类实例。实现接口的 createMessage() 方法我们可以看出，主要目的是根据 session 创建用户自定义的消息内容。

继续查看 send() 方法中内容。

/**
 * 发送消息
 * @param destination the destination to send this message to
 * @param messageCreator callback to create a message
 * @throws JmsException
 */
@Override
public void send(final Destination destination, final MessageCreator messageCreator) throws JmsException {
   execute(session -> {
      doSend(session, destination, messageCreator);
      return null;
   }, false);
}

看到 send() 方法中又调用了 execute() 方法，我们不得不回想起 JdbcTemplate 实现风格，极为相似，两者都是提取一个公共的方法作为最底层、最通用的功能实现，然后又通过回调函数的不同来区分个性化的功能。我们首先查看通用代码的抽取实现。

① 通用代码抽取

通过 send() 方法可以看出，需要传入的参数包含会话 SessionCallback 回调函数接口实现以及一个布尔值表示是否开启向服务器推送连接信息，只有接收信息时需要，发送不需要。由于Spring5对于函数式接口，都采用了 lambda 表达式写法，所以看起来有点不够清晰，其实本质上就是实现 SessionCallback 接口的 doInJms() 方法。

execute(new SessionCallback<Object>() {
    public Object doInJms(Session session) throws JMSException {
        doSend(session, destination, messageCreator);
        return null;
    }
}, false);

也就是说，doInJms() 方法中最后真正做的是 doSend(session, destination, messageCreator) 方法，就是实际发送消息的操作，这其实仅与发送有关，我们完全可以把 doInJms() 方法的实际内容替换成接收消息方法，所以Spring利用回调接口，先进行通用代码的抽取。

我们回过头来研究 execute() 方法，没错，execute() 方法就是通用代码部分。根据之前分析 JdbcTemplate 的经验，我们推断，在 execute() 中一定是封装了 Connection 以及 Session 的创建操作等套路代码。

/**
 * Execute the action specified by the given action object within a
 * JMS Session. Generalized version of {@code execute(SessionCallback)},
 * allowing the JMS Connection to be started on the fly.
 * <p>Use {@code execute(SessionCallback)} for the general case.
 * Starting the JMS Connection is just necessary for receiving messages,
 * which is preferably achieved through the {@code receive} methods.
 * @param action callback object that exposes the Session
 * @param startConnection whether to start the Connection
 * @return the result object from working with the Session
 * @throws JmsException if there is any problem
 * @see #execute(SessionCallback)
 * @see #receive
 */
@Nullable
public <T> T execute(SessionCallback<T> action, boolean startConnection) throws JmsException {
   Assert.notNull(action, "Callback object must not be null");
   Connection conToClose = null;
   Session sessionToClose = null;
   try {
      // 尝试获取session
      Session sessionToUse = ConnectionFactoryUtils.doGetTransactionalSession(
            obtainConnectionFactory(), this.transactionalResourceFactory, startConnection);
      if (sessionToUse == null) {
         // 创建connection
         conToClose = createConnection();
         // 根据connection创建session
         sessionToClose = createSession(conToClose);
         // 是否开启向服务器推送连接信息，只有接收信息时需要，发送不需要
         if (startConnection) {
            conToClose.start();
         }
         sessionToUse = sessionToClose;
      }
      if (logger.isDebugEnabled()) {
         logger.debug("Executing callback on JMS Session: " + sessionToUse);
      }
      // 调用回调方法
      return action.doInJms(sessionToUse);
   }
   catch (JMSException ex) {
      throw convertJmsAccessException(ex);
   }
   finally {
      // 关闭连接
      JmsUtils.closeSession(sessionToClose);
      // 释放连接
      ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), startConnection);
   }
}

在展示单独使用 ActiveMQ 时，我们知道为了发送一条消息需要做很多工作，需要很多的辅助代码，而这些代码又都是千篇一律的，没有任何的差异，所以 execute() 方法的目的就是帮助我们抽离这些冗余代码使我们更加专注于业务逻辑的实现。从函数中看，这些冗余代码包括创建 Connection、创建 Session、当然也包括关闭 Session 和关闭 Connection。而在准备工作结束后，调用回调函数将程序引入用户自定义实现的个性化处理。

简单的看一下创建 Connection、创建 Session、关闭 Session 和关闭 Connection 的实现，可以看到和我们单独使用 ActiveMQ 的代码本质上是一致的。

/**
 * Create a JMS Connection via this template's ConnectionFactory.
 * <p>This implementation uses JMS 1.1 API.
 * @return the new JMS Connection
 * @throws JMSException if thrown by JMS API methods
 * @see javax.jms.ConnectionFactory#createConnection()
 */
protected Connection createConnection() throws JMSException {
   return obtainConnectionFactory().createConnection();
}

/**
 * Create a JMS Session for the given Connection.
 * <p>This implementation uses JMS 1.1 API.
 * @param con the JMS Connection to create a Session for
 * @return the new JMS Session
 * @throws JMSException if thrown by JMS API methods
 * @see javax.jms.Connection#createSession(boolean, int)
 */
protected Session createSession(Connection con) throws JMSException {
    return con.createSession(isSessionTransacted(), getSessionAcknowledgeMode());
}

/**
 * Close the given JMS Session and ignore any thrown exception.
 * This is useful for typical {@code finally} blocks in manual JMS code.
 * @param session the JMS Session to close (may be {@code null})
 */
public static void closeSession(@Nullable Session session) {
    if (session != null) {
        try {
            session.close();
        }
        catch (JMSException ex) {
            logger.trace("Could not close JMS Session", ex);
        }
        catch (Throwable ex) {
            // We don't trust the JMS provider: It might throw RuntimeException or Error.
            logger.trace("Unexpected exception on closing JMS Session", ex);
        }
    }
}

/**
 * Release the given Connection, stopping it (if necessary) and eventually closing it.
 * <p>Checks {@link SmartConnectionFactory#shouldStop}, if available.
 * This is essentially a more sophisticated version of
 * {@link org.springframework.jms.support.JmsUtils#closeConnection}.
 * @param con the Connection to release
 * (if this is {@code null}, the call will be ignored)
 * @param cf the ConnectionFactory that the Connection was obtained from
 * (may be {@code null})
 * @param started whether the Connection might have been started by the application
 * @see SmartConnectionFactory#shouldStop
 * @see org.springframework.jms.support.JmsUtils#closeConnection
 */
public static void releaseConnection(@Nullable Connection con, @Nullable ConnectionFactory cf, boolean started) {
    if (con == null) {
        return;
    }
    if (started && cf instanceof SmartConnectionFactory && ((SmartConnectionFactory) cf).shouldStop(con)) {
        try {
            con.stop();
        }
        catch (Throwable ex) {
            logger.debug("Could not stop JMS Connection before closing it", ex);
        }
    }
    try {
        con.close();
    }
    catch (Throwable ex) {
        logger.debug("Could not close JMS Connection", ex);
    }
}

② 发送消息的实现

当然，在开启关闭资源中间， execute() 方法调用了回调函数的回调接口方法 doInJms(sessionToUse)，传入了刚刚创建的会话 session 参数。有了基类辅助实现，使Spring更加专注于个性的处理，也就是说Spring使用 execute() 方法中封装了冗余代码，而将个性化的代码实现放在了回调函数 doInJms(sessionToUse)函数中。在发送消息的功能中回调函数通过局部类实现。

execute(session -> {
    doSend(session, destination, messageCreator);
    return null;
}, false);

这样发送消息的逻辑就转为 doSend() 方法中，查看该方法内容，应该主要实现利用 session、destination 和 messageCreator 发送消息逻辑。

/**
 * 发送JMS消息
 * Send the given JMS message.
 * @param session the JMS Session to operate on
 * @param destination the JMS Destination to send to
 * @param messageCreator callback to create a JMS Message
 * @throws JMSException if thrown by JMS API methods
 */
protected void doSend(Session session, Destination destination, MessageCreator messageCreator)
      throws JMSException {

   Assert.notNull(messageCreator, "MessageCreator must not be null");
   // 根据destination创建MessageProducer(通用方法)
   MessageProducer producer = createProducer(session, destination);
   try {
      // 创建消息内容(个性化内容需要调用用户自己实现的方法)
      Message message = messageCreator.createMessage(session);
      if (logger.isDebugEnabled()) {
         logger.debug("Sending created message: " + message);
      }
      // 利用MessageProducer发送消息
      doSend(producer, message);
      // Check commit - avoid commit call within a JTA transaction.
      if (session.getTransacted() && isSessionLocallyTransacted(session)) {
         // Transacted session created by this template -> commit.
         JmsUtils.commitIfNecessary(session);
      }
   }
   finally {
      JmsUtils.closeMessageProducer(producer);
   }
}

/**
 * Actually send the given JMS message.
 * @param producer the JMS MessageProducer to send with
 * @param message the JMS Message to send
 * @throws JMSException if thrown by JMS API methods
 */
protected void doSend(MessageProducer producer, Message message) throws JMSException {
    if (this.deliveryDelay >= 0) {
        producer.setDeliveryDelay(this.deliveryDelay);
    }
    if (isExplicitQosEnabled()) {
        producer.send(message, getDeliveryMode(), getPriority(), getTimeToLive());
    }
    else {
        producer.send(message);
    }
}

messageCreator 在之前保存了我们想要发送的消息内容，调用我们实现的 createMessage() 接口方法即可获取消息。在演示独立使用消息功能的时候，我们大体了解了消息发送的基本套路，虽然这些步骤已经被Spring拆得支离破碎，但是我们还是能捕捉到一些影子。在发送消息还是遵循着消息发送的规则，比如根据 Destination 创建 MessageProducer、创建 Message ，并使用 MessageProducer 实例来发送消息。

③ 接收消息的实现

有了发送消息的分析基础，我们再来查看接收方法的实现逻辑。我们通常使用 jmsTemplate.receive(destination) 来接收简单的消息，那么这个功能Spring是如何封装的呢？

/**
 * 接收消息
 * @param destination the destination to receive a message from
 * @return
 * @throws JmsException
 */
@Override
@Nullable
public Message receive(Destination destination) throws JmsException {
    return receiveSelected(destination, null);
}

@Override
@Nullable
public Message receiveSelected(final Destination destination, @Nullable final String messageSelector) throws JmsException {
    return execute(session -> doReceive(session, destination, messageSelector), true);
}

可以看出，接收方法同样调用了通用方法 execute()，第二个参数传递的为 true，意味着需要向服务器推送连接信息。

我们主要需要查看的是回调接口的实现方法中的 doReceive() 内容。

/**
 * 接收JMS消息
 * Receive a JMS message.
 * @param session the JMS Session to operate on
 * @param destination the JMS Destination to receive from
 * @param messageSelector the message selector for this consumer (can be {@code null})
 * @return the JMS Message received, or {@code null} if none
 * @throws JMSException if thrown by JMS API methods
 */
@Nullable
protected Message doReceive(Session session, Destination destination, @Nullable String messageSelector)
      throws JMSException {

   return doReceive(session, createConsumer(session, destination, messageSelector));
}

/**
 * 接收JMS消息
 * Actually receive a JMS message.
 * @param session the JMS Session to operate on
 * @param consumer the JMS MessageConsumer to receive with
 * @return the JMS Message received, or {@code null} if none
 * @throws JMSException if thrown by JMS API methods
 */
@Nullable
protected Message doReceive(Session session, MessageConsumer consumer) throws JMSException {
    try {
        // Use transaction timeout (if available).
        long timeout = getReceiveTimeout();
        // 获取connectionFactory
        ConnectionFactory connectionFactory = getConnectionFactory();
        JmsResourceHolder resourceHolder = null;
        // 从connectionFactory中获取连接，并包装成JmsResourceHolder返回
        if (connectionFactory != null) {
            resourceHolder = (JmsResourceHolder) TransactionSynchronizationManager.getResource(connectionFactory);
        }
        if (resourceHolder != null && resourceHolder.hasTimeout()) {
            timeout = Math.min(timeout, resourceHolder.getTimeToLiveInMillis());
        }
        // 接收消息
        Message message = receiveFromConsumer(consumer, timeout);
        if (session.getTransacted()) {
            // Commit necessary - but avoid commit call within a JTA transaction.
            if (isSessionLocallyTransacted(session)) {
                // Transacted session created by this template -> commit.
                JmsUtils.commitIfNecessary(session);
            }
        }
        else if (isClientAcknowledge(session)) {
            // Manually acknowledge message, if any.
            if (message != null) {
                message.acknowledge();
            }
        }
        return message;
    }
    finally {
        JmsUtils.closeMessageConsumer(consumer);
    }
}

/**
 * Actually receive a message from the given consumer.
 * @param consumer the JMS MessageConsumer to receive with
 * @param timeout the receive timeout (a negative value indicates
 * a no-wait receive; 0 indicates an indefinite wait attempt)
 * @return the JMS Message received, or {@code null} if none
 * @throws JMSException if thrown by JMS API methods
 * @since 4.3
 * @see #RECEIVE_TIMEOUT_NO_WAIT
 * @see #RECEIVE_TIMEOUT_INDEFINITE_WAIT
 */
@Nullable
protected Message receiveFromConsumer(MessageConsumer consumer, long timeout) throws JMSException {
    if (timeout > 0) {
        return consumer.receive(timeout);
    }
    else if (timeout < 0) {
        return consumer.receiveNoWait();
    }
    else {
        return consumer.receive();
    }
}

实现的套路与发送差不多，同样还是使用 execute() 函数来封装冗余的公共操作，而最终的目标还是通过 consumer.receive() 来接收消息，其中的过程就是对于 MessageConsumer 的创建以及一些辅助操作。

消息监听器

消息监听器容器是一个用于查看 JMS 目标等待消息到达的特殊 bean，一旦消息到达它就可以获取到消息，并通过调用 onMessage() 方法将消息传递给一个MessageListener 实现。

Spring中消息监听器容器的类型主要有：

SimpleMessageListenerContainer：最简单的消息监听器容器，只能处理固定数量的JMS会话，且不支持事务。
DefaultMessageListenerContainer：这个消息监听器容器建立在 SimpleMessageListenerContainer 容器之上，添加了对事务的支持。

我们开头Spring整合ActiveMQ的示例中使用的是 DefaultMessageListenerContainer 消息监听器容器，以此为例进行分析消息监听器容器的实现。

通过之前的配置文件的定义，可以看到我们将自定义的消息监听器注入进监听器容器中，这样才可以在收到信息时，容器把消息转向监听器处理。

<bean id="myMessageListener" class="guo.ping.activemq.MyMessageListener" />

<bean id="javaConsumer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    ····
    <property name="messageListener" ref="myMessageListener" />
</bean>

我们查看 DefaultMessageListenerContainer 类的继承结构图，同样发现其顶端实现了 InitializingBean 接口。

Spring源码——消息队列

我们还是首先查看接口方法 afterPropertiesSet() 中的逻辑，其方法实现在其父类 AbstractJmsListeningContainer 中。

/**
 * 实现自InitializingBean接口
 * Delegates to {@link #validateConfiguration()} and {@link #initialize()}.
 */
@Override
public void afterPropertiesSet() {
   // 父类对connectionFactory判空
   super.afterPropertiesSet();
   // 验证配置文件
   validateConfiguration();
   // 初始化容器
   initialize();
}

监听器容器的初始化只包含了三句代码，第一句调用父类的 afterPropertiesSet() 方法，其实就是 JmsAccessor 抽象类中的实现，仅仅是对 ConnectionFactory 实例判空。

/**
 * 实现InitializingBean接口
 */
@Override
public void afterPropertiesSet() {
   // 对ConnectionFactory判空
   if (getConnectionFactory() == null) {
      throw new IllegalArgumentException("Property 'connectionFactory' is required");
   }
}

第二句代码为对配置进行验证，在 AbstractJmsListeningContainer 中仅仅定义了一个空模板方法，交由子类去重写完成校验逻辑。

/**
 * 模板方法
 * Validate the configuration of this container.
 * <p>The default implementation is empty. To be overridden in subclasses.
 */
protected void validateConfiguration() {
}

重写方法在 AbstractMessageListenerContainer 类 (UML图) 中实现。主要逻辑其实就是对 destination 判空。

/**
 * 实现AbstractJmsListeningContainer的模板方法，对destination判空
 */
@Override
protected void validateConfiguration() {
   if (this.destination == null) {
      throw new IllegalArgumentException("Property 'destination' or 'destinationName' is required");
   }
}

第三句代码是进行初始化操作，委托 initialize() 方法进行执行。initialize() 方法其实在三个类中分别进行了实现，所以我们真实调用 initialize() 方法必然是从 DefaultMessageListenerContainer 中进入的。

Spring源码——消息队列

① 初始化容器

先来看 DefaultMessageListenerContainer 中的 initialize() 方法。

/**
 * 实现父类的initialize()方法，在afterPropertiesSet()中调用
 */
@Override
public void initialize() {
   // 如果是默认的，那么分有事务还是没事务，有事务就是不缓存，没有事务就是CACHE_CONSUMER
   if (this.cacheLevel == CACHE_AUTO) {
      this.cacheLevel = (getTransactionManager() != null ? CACHE_NONE : CACHE_CONSUMER);
   }

   // Prepare taskExecutor and maxMessagesPerTask.
   synchronized (this.lifecycleMonitor) {
      if (this.taskExecutor == null) {
         // 创建默认的任务执行器taskExecutor
         this.taskExecutor = createDefaultTaskExecutor();
      }
      else if (this.taskExecutor instanceof SchedulingTaskExecutor &&
            ((SchedulingTaskExecutor) this.taskExecutor).prefersShortLivedTasks() &&
            this.maxMessagesPerTask == Integer.MIN_VALUE) {
         // TaskExecutor indicated a preference for short-lived tasks. According to
         // setMaxMessagesPerTask javadoc, we'll use 10 message per task in this case
         // unless the user specified a custom value.
         this.maxMessagesPerTask = 10;
      }
   }

   // Proceed with actual listener initialization.
   super.initialize();
}

首先方法中对于 this.cacheLevel 的值进行了确定，这个值关系到 DefaultMessageListenerContainer 缓不缓存 connection，session 和 consumer。发送消息时我们可以使用 PooledConnectionFactory 或者 CachingConnectionFactory，用来缓存JMS资源，具体参考深入了解PooledConnectionFactory 与CachingConnectionFactory。接收消息则是根据 this.cacheLevel 属性来决定是否缓存 connection，session 和 consumer。

this.cacheLevel 属性设置不同的值含义如下：

常量	值	含义
CATCH_NONE	0	不缓存 JMS 任何资源
CATCH_CONNECTION	1	只缓存 JMS 的共享 Connection
CATCH_SESSION	2	缓存 JMS 的共享 Connection和 Session
CATCH_CONSUMER	3	缓存 JMS 的共享 Connection 和 Session 还有 MessageConsumer
CATCH_AUTO	4	系统根据事务管理策略自动选择一个合适的 cacheLevel，如果存在事务，则为CATCH_NONE，不存在事务为CACHE_CONSUMER。

可以看到，Spring对于 CATCH_AUTO 的值判断就是在这里进行的。

DefaultMessageListenerContainer 其实是一个用于异步消息监听的管理类，其核心是包含一个任务执行器，不断的执行任务——消息监听。而这个任务执行器其实就是内部初始化建立的一个 taskExecutor，也就是这里的 this.taskExecutor。任务执行器的创建就是这里进行的，调用了 createDefaultTaskExecutor() 方法。

/**
 * 创建一个默认的SimpleAsyncTaskExecutor
 * SimpleAsyncTaskExecutor不重用任何线程，或者说它每次调用都启动一个新线程。但是，它还是支持对并发总数设限，
 * 当超过线程并发总数限制时，阻塞新的调用，直到有位置被释放。
 * Create a default TaskExecutor. Called if no explicit TaskExecutor has been specified.
 * <p>The default implementation builds a {@link org.springframework.core.task.SimpleAsyncTaskExecutor}
 * with the specified bean name (or the class name, if no bean name specified) as thread name prefix.
 * @see org.springframework.core.task.SimpleAsyncTaskExecutor#SimpleAsyncTaskExecutor(String)
 */
protected TaskExecutor createDefaultTaskExecutor() {
    String beanName = getBeanName();
    String threadNamePrefix = (beanName != null ? beanName + "-" : DEFAULT_THREAD_NAME_PREFIX);
    return new SimpleAsyncTaskExecutor(threadNamePrefix);
}

这里默认的任务执行器类型是 SimpleAsyncTaskExecutor，这个执行器的缺点是不会重用连接，也就是对于每个任务都需要新开启一个线程，执行完任务后会关闭它。如果要优化的话可以考虑使用线程池。但是，它还是支持对并发总数设限，当超过线程并发总数限制时，阻塞新的调用，直到有位置才被释放。

做完这两件事 (cacheLevel 和 taskExecutor) 之后，调用父类 AbstractPollingMessageListenerContainer 的 initialize() 方法。

@Override
public void initialize() {
   // Set sessionTransacted=true in case of a non-JTA transaction manager.
   if (!this.sessionTransactedCalled &&
         this.transactionManager instanceof ResourceTransactionManager &&
         !TransactionSynchronizationUtils.sameResourceFactory(
               (ResourceTransactionManager) this.transactionManager, obtainConnectionFactory())) {
      super.setSessionTransacted(true);
   }

   // Use bean name as default transaction name.
   if (this.transactionDefinition.getName() == null) {
      String beanName = getBeanName();
      if (beanName != null) {
         this.transactionDefinition.setName(beanName);
      }
   }

   // Proceed with superclass initialization.
   super.initialize();
}

主要包含对 sessionTransacted 属性的设置以及事务属性中名称的设置。随后继续调用 AbstractJmsListeningContainer 类中的 initialize() 方法。

/**
 * 初始化容器
 * Initialize this container.
 * <p>Creates a JMS Connection, starts the {@link javax.jms.Connection}
 * (if {@link #setAutoStartup(boolean) "autoStartup"} hasn't been turned off),
 * and calls {@link #doInitialize()}.
 * @throws org.springframework.jms.JmsException if startup failed
 */
public void initialize() throws JmsException {
   try {
      // lifecycleMonitor用于控制生命周期的同步处理
      synchronized (this.lifecycleMonitor) {
         // 将active置为true
         this.active = true;
         this.lifecycleMonitor.notifyAll();
      }
      doInitialize();
   }
   catch (JMSException ex) {
      synchronized (this.sharedConnectionMonitor) {
         ConnectionFactoryUtils.releaseConnection(this.sharedConnection, getConnectionFactory(), this.autoStartup);
         this.sharedConnection = null;
      }
      throw convertJmsAccessException(ex);
   }
}

初始化先将 this.active 设置为 true，并将获取到 this.lifecycleMonitor 对象锁，之后又进入等待 (Waiting) 状态的线程全部唤醒。之后又执行 doInitialize() 方法，该方法在 AbstractJmsListeningContainer 类仅仅是抽象方法，需要子类重写，重写是在 DefaultMessageListenerContainer 中完成的。

/**
 * 消息监听器允许创建多个Session和MessageConsumer来接收消息，具体的个数由该属性指定
 */
private int concurrentConsumers = 1;

/**
 * 实现AbstractJmsListeningContainer的模板方法，初始化
 * Creates the specified number of concurrent consumers,
 * in the form of a JMS Session plus associated MessageConsumer
 * running in a separate thread.
 * @see #scheduleNewInvoker
 * @see #setTaskExecutor
 */
@Override
protected void doInitialize() throws JMSException {
   synchronized (this.lifecycleMonitor) {
      for (int i = 0; i < this.concurrentConsumers; i++) {
         scheduleNewInvoker();
      }
   }
}

方法中利用 this.lifecycleMonitor 锁对象实现方法同步，主要根据 this.concurrentConsumers 值不断的调用 scheduleNewInvoker() 方法。消息监听器允许创建多个 Session 和 MessageConsumer 来接收消息，具体的个数由 concurrentConsumers 属性指定。那么可以猜测 scheduleNewInvoker() 方法主要是创建监听消息的任务，让之前创建的任务执行器 taskExecutor 去执行。

需要注意的是，应该只是在 Destination 为 Queue 的时候才使用多个 MessageConsumer (Queue 中的一个消息只能被一个 Consumer 接收)，虽然使用多个 MessageConsumer 会提高消息处理的性能，但是消息处理的顺序却得不到保证。消息被接收的顺序仍然是消息发送时的顺序，但是由于消息可能会被并发处理，因此消息处理的顺序可能和消息发送的顺序不同。此外，不应该在 Destination 为 Topic 的时候使用多个 MessageConsumer，因为多个 MessageConsumer 会接收到同样的消息。这其实就是多线程带来的问题。

下面接着研究 scheduleNewInvoker() 方法。

/**
 * 任务集合。
 * 事实上一个消费者对应了一个AsyncMessageListenerInvoker任务，每个任务需要一个单独的线程去执行它。
 * 这个AsyncMessageListenerInvoker实例被放在了该集合中
 */
private final Set<AsyncMessageListenerInvoker> scheduledInvokers = new HashSet<>();

/**
 * 创建一个新的调用线程
 * Schedule a new invoker, increasing the total number of scheduled
 * invokers for this listener container.
 */
private void scheduleNewInvoker() {
   // 创建一个监听消息的任务
   AsyncMessageListenerInvoker invoker = new AsyncMessageListenerInvoker();
   if (rescheduleTaskIfNecessary(invoker)) {
      // This should always be true, since we're only calling this when active.
      this.scheduledInvokers.add(invoker);
   }
}

之前我们已经说了消息监听器容器本质是创建了任务执行器 taskExecutor，为了执行任务——消息监听。消息监听任务其实就是一个 Runnable 接口的实现类。Spring将消息监听的任务被抽象成 AsyncMessageListenerInvoker 类，这个类实现了 Runnable 接口，内部 run() 方法其实是通过不断循环 MessageConsumer.receive() 方法来实现一直监听。

我们已经道出了消息监听器的本质，本文的剩余部分就是着重查看源代码解释上面这一段话。

/**
 * 消息监听任务，本质是Runnable
 * Runnable that performs looped {@code MessageConsumer.receive()} calls.
 */
private class AsyncMessageListenerInvoker implements SchedulingAwareRunnable {
	@Override
    public void run() {
		···
    }
}

scheduleNewInvoker() 方法中首先创建了一个监听消息的任务，并加入到 this.scheduledInvokers 集合中。由于该方法之前是在 for 循环中使用的，循环次数是 this.concurrentConsumers 值，所以其实就是一个消费者对应了一个 AsyncMessageListenerInvoker 任务，每个任务需要一个单独的线程去执行它。这个 AsyncMessageListenerInvoker 实例被放在了 this.scheduledInvokers 集合里面。当然，在加入集合之前，还调用了 rescheduleTaskIfNecessary(invoker) 方法进行了判断，那么这个方法是什么意思呢？

/**
 * Take the given task object and reschedule it, either immediately if
 * this container is currently running, or later once this container
 * has been restarted.
 * <p>If this container has already been shut down, the task will not
 * get rescheduled at all.
 * @param task the task object to reschedule
 * @return whether the task has been rescheduled
 * (either immediately or for a restart of this container)
 * @see #doRescheduleTask
 */
protected final boolean rescheduleTaskIfNecessary(Object task) {
   if (this.running) {
      try {
         doRescheduleTask(task);
      }
      catch (RuntimeException ex) {
         logRejectedTask(task, ex);
         this.pausedTasks.add(task);
      }
      return true;
   }
   else if (this.active) {
      this.pausedTasks.add(task);
      return true;
   }
   else {
      return false;
   }
}

/**
 * 任务执行器执行任务
 * Re-executes the given task via this listener container's TaskExecutor.
 * @see #setTaskExecutor
 */
@Override
protected void doRescheduleTask(Object task) {
    Assert.state(this.taskExecutor != null, "No TaskExecutor available");
    this.taskExecutor.execute((Runnable) task);
}

doRescheduleTask() 函数其实是任务执行器执行任务开启一个线程执行任务，查看 SimpleAsyncTaskExecutor 的 execute() 方法，其实质调用了 doExecute() 方法。

/**
 * Template method for the actual execution of a task.
 * <p>The default implementation creates a new Thread and starts it.
 * @param task the Runnable to execute
 * @see #setThreadFactory
 * @see #createThread
 * @see java.lang.Thread#start()
 */
protected void doExecute(Runnable task) {
   // 开启一个线程执行任务
   Thread thread = (this.threadFactory != null ? this.threadFactory.newThread(task) : createThread(task));
   thread.start();
}

回头我们再分析 rescheduleTaskIfNecessary() 方法，其实我们并没有调用 doRescheduleTask(task) 方法立即执行任务，因为判断标记 this.running 还是 false，而 this.active 之前我们设置为 true，所以会将创建出的消息监听任务全部放入 this.pausedTasks 集合中。

/**
 * 存储暂停的任务
 */
private final List<Object> pausedTasks = new LinkedList<>();

② 启动消息监听任务

之前我们仅仅只是相当于创建了消息监听的容器，容器创建了任务执行器以及要执行的任务，但是并没有执行，任务都被加入到 this.pausedTasks 集合中，我们在测试代码中也没有发现明显启动线程执行的方法，那么问题来了，Spring究竟如何启动任务执行器执行任务的呢？

其实答案在 ApplicationContext 中，之前分析 ApplicationContext 的执行流程时，在 refresh() 方法的最后一步是 finishRefresh() 方法。

// Last step: publish corresponding event.
// 完成刷新过程，通知生命周期处理器 lifecycleProcessor 刷新过程，同时发出 ContextRefreshEvent 通知别人
finishRefresh();

看方法的注释，其中包含通知生命周期处理器 lifecycleProcessor 刷新过程。生命周期处理器又是什么呢？

Lifecycle 接口定义了容器中每个对象的重要方法，每个对象都有自己的生命周期需求。

public interface Lifecycle {
    void start();
    void stop();
    boolean isRunning();
}

任何Spring管理的对象可能实现这个接口，当 ApplicationContext 自身启动和停止时，它将自动调用上下文内所有生命周期的实现。通过委托给 LifecycleProcessor 来做这个工作。LifecycleProcessor 自身扩展了 Lifecycle 接口，同时增加了两个其他的方法来与上下文交互，使得其可以刷新和关闭。`

public interface LifecycleProcessor extends Lifecycle {
    void onRefresh();
    void onClose();
}

那么 ApplicationContext 自身启动和停止时，如何自动调用上下文内所有生命周期的实现的呢？查看 finishRefresh() 方法内容。

/**
 * 完成刷新过程，通知生命周期处理器 lifecycleProcessor 刷新过程，同时发出 ContextRefreshEvent 通知别人
 * Finish the refresh of this context, invoking the LifecycleProcessor's
 * onRefresh() method and publishing the
 * {@link org.springframework.context.event.ContextRefreshedEvent}.
 */
protected void finishRefresh() {
   // Clear context-level resource caches (such as ASM metadata from scanning).
   clearResourceCaches();

   // Initialize lifecycle processor for this context.
   // 当ApplicationContext启动或者停止的时候，它会通过LifecycleProcessor来与所有声明的bean的周期做状态更新，而在使用前先初始化
   initLifecycleProcessor();

   // Propagate refresh to lifecycle processor first.
   // 启动所有实现了Lifecycle接口的bean
   getLifecycleProcessor().onRefresh();

   // Publish the final event.
   // 当完成ApplicationContext初始化时，要通过Spring中的事件发布机制来发出ContextRefreshedEvent事件，让监听器进一步处理
   publishEvent(new ContextRefreshedEvent(this));

   // Participate in LiveBeansView MBean, if active.
   LiveBeansView.registerApplicationContext(this);
}

可以看到其中有一个 getLifecycleProcessor().onRefresh() 方法，ApplicationContext 启动所有实现了 Lifecycle 接口的 bean。

首先先获得实现 LifecycleProcessor 接口的 DefaultLifecycleProcessor 实例。

/**
 * 返回DefaultLifecycleProcessor
 * Return the internal LifecycleProcessor used by the context.
 * @return the internal LifecycleProcessor (never {@code null})
 * @throws IllegalStateException if the context has not been initialized yet
 */
LifecycleProcessor getLifecycleProcessor() throws IllegalStateException {
   if (this.lifecycleProcessor == null) {
      throw new IllegalStateException("LifecycleProcessor not initialized - " +
            "call 'refresh' before invoking lifecycle methods via the context: " + this);
   }
   return this.lifecycleProcessor;
}

然后调用 DefaultLifecycleProcessor 的 onRefresh() 方法。

@Override
public void onRefresh() {
   startBeans(true);
   this.running = true;
}

private void startBeans(boolean autoStartupOnly) {
    Map<String, Lifecycle> lifecycleBeans = getLifecycleBeans();
    Map<Integer, LifecycleGroup> phases = new HashMap<>();
    lifecycleBeans.forEach((beanName, bean) -> {
        if (!autoStartupOnly || (bean instanceof SmartLifecycle && ((SmartLifecycle) bean).isAutoStartup())) {
            int phase = getPhase(bean);
            LifecycleGroup group = phases.get(phase);
            if (group == null) {
                group = new LifecycleGroup(phase, this.timeoutPerShutdownPhase, lifecycleBeans, autoStartupOnly);
                phases.put(phase, group);
            }
            group.add(beanName, bean);
        }
    });
    if (!phases.isEmpty()) {
        List<Integer> keys = new ArrayList<>(phases.keySet());
        Collections.sort(keys);
        for (Integer key : keys) {
            phases.get(key).start();
        }
    }
}

那么这些方法和我们要说的消息监听器容器有什么关系呢？我们再重新看一下 DefaultMessageListenerContainer 的类继承结构，这次我们发现它还实现了 Lifecycle 接口。

Spring源码——消息队列

所以在 startBeans() 方法中，通过 getLifecycleBeans() 方法就能够获得 DefaultMessageListenerContainer 的 bean。

Spring源码——消息队列

经过层层调用最终执行了监听器容器 DefaultMessageListenerContainer 的 start() 方法。

/**
 * Overridden to reset the stop callback, if any.
 */
@Override
public void start() throws JmsException {
   synchronized (this.lifecycleMonitor) {
      this.stopCallback = null;
   }
   super.start();
}

查看 AbstractJmsListeningContainer 中的方法。

/**
 * 启动任务执行器执行任务
 * Start this container.
 * @throws JmsException if starting failed
 * @see #doStart
 */
@Override
public void start() throws JmsException {
    try {
        doStart();
    }
    catch (JMSException ex) {
        throw convertJmsAccessException(ex);
    }
}

/**
 * Start the shared Connection, if any, and notify all invoker tasks.
 * @throws JMSException if thrown by JMS API methods
 * @see #startSharedConnection
 */
protected void doStart() throws JMSException {
    // Lazily establish a shared Connection, if necessary.
    if (sharedConnectionEnabled()) {
        establishSharedConnection();
    }

    // Reschedule paused tasks, if any.
    synchronized (this.lifecycleMonitor) {
        this.running = true;
        this.lifecycleMonitor.notifyAll();
        resumePausedTasks();
    }

    // Start the shared Connection, if any.
    if (sharedConnectionEnabled()) {
        startSharedConnection();
    }
}

查看 doStart() 方法逻辑，首先是建立一个容器内共享的连接。

@Override
protected final boolean sharedConnectionEnabled() {
   return (getCacheLevel() >= CACHE_CONNECTION);
}

/**
 * Overridden to accept a failure in the initial setup - leaving it up to the
 * asynchronous invokers to establish the shared Connection on first access.
 * @see #refreshConnectionUntilSuccessful()
 */
@Override
protected void establishSharedConnection() {
    try {
        super.establishSharedConnection();
    }
    catch (Exception ex) {
        if (ex instanceof JMSException) {
            invokeExceptionListener((JMSException) ex);
        }
        logger.debug("Could not establish shared JMS Connection - " +
                     "leaving it up to asynchronous invokers to establish a Connection as soon as possible", ex);
    }
}

/**
 * Establish a shared Connection for this container.
 * <p>The default implementation delegates to {@link #createSharedConnection()},
 * which does one immediate attempt and throws an exception if it fails.
 * Can be overridden to have a recovery process in place, retrying
 * until a Connection can be successfully established.
 * @throws JMSException if thrown by JMS API methods
 */
protected void establishSharedConnection() throws JMSException {
    synchronized (this.sharedConnectionMonitor) {
        if (this.sharedConnection == null) {
            this.sharedConnection = createSharedConnection();
            logger.debug("Established shared JMS Connection");
        }
    }
}

之后我们明显的看到将 this.running 标记位置为 true，this.lifecycleMonitor 唤醒了所有等待线程，通过 resumePausedTasks() 恢复/启动所有的暂停任务。查看 resumePausedTasks() 方法。

/**
 * Try to resume all paused tasks.
 * Tasks for which rescheduling failed simply remain in paused mode.
 */
protected void resumePausedTasks() {
   synchronized (this.lifecycleMonitor) {
      if (!this.pausedTasks.isEmpty()) {
         for (Iterator<?> it = this.pausedTasks.iterator(); it.hasNext();) {
            Object task = it.next();
            try {
               // 执行任务
               doRescheduleTask(task);
               it.remove();
               if (logger.isDebugEnabled()) {
                  logger.debug("Resumed paused task: " + task);
               }
            }
            catch (RuntimeException ex) {
               logRejectedTask(task, ex);
               // Keep the task in paused mode...
            }
         }
      }
   }
}

显而易见，resumePausedTasks() 方法中对暂停任务集合 this.pausedTasks 进行遍历，然后调用之前分析过的 doRescheduleTask(task) 方法，不过这一次 this.running 标记位置已经为 true，也就是正式执行了。

③ 监听任务实现细节

消息监听的任务被抽象成 AsyncMessageListenerInvoker 类，所以我们重点查看该类的 run() 方法是如何实现的。

@Override
public void run() {
    // 并发控制
    synchronized (lifecycleMonitor) {
        activeInvokerCount++;
        // 唤醒所有在lifecycleMonitor对象锁上的线程
        lifecycleMonitor.notifyAll();
    }
    boolean messageReceived = false;
    try {
        // 根据每个任务设置的最大处理消息数量而作不同处理
        // 小于0默认为无限制，一直接收消息
        if (maxMessagesPerTask < 0) {
            messageReceived = executeOngoingLoop();
        }
        else {
            int messageCount = 0;
            // 消息数量控制，一旦超出数量则停止循环
            while (isRunning() && messageCount < maxMessagesPerTask) {
                messageReceived = (invokeListener() || messageReceived);
                messageCount++;
            }
        }
    }
    catch (Throwable ex) {
        // 清理操作，包括关闭session等
        clearResources();
        if (!this.lastMessageSucceeded) {
            // We failed more than once in a row or on startup -
            // wait before first recovery attempt.
            waitBeforeRecoveryAttempt();
        }
        this.lastMessageSucceeded = false;
        boolean alreadyRecovered = false;
        synchronized (recoveryMonitor) {
            if (this.lastRecoveryMarker == currentRecoveryMarker) {
                handleListenerSetupFailure(ex, false);
                recoverAfterListenerSetupFailure();
                currentRecoveryMarker = new Object();
            }
            else {
                alreadyRecovered = true;
            }
        }
        if (alreadyRecovered) {
            handleListenerSetupFailure(ex, true);
        }
    }
    finally {
        synchronized (lifecycleMonitor) {
            decreaseActiveInvokerCount();
            lifecycleMonitor.notifyAll();
        }
        if (!messageReceived) {
            this.idleTaskExecutionCount++;
        }
        else {
            this.idleTaskExecutionCount = 0;
        }
        synchronized (lifecycleMonitor) {
            if (!shouldRescheduleInvoker(this.idleTaskExecutionCount) || !rescheduleTaskIfNecessary(this)) {
                // We're shutting down completely.
                scheduledInvokers.remove(this);
                if (logger.isDebugEnabled()) {
                    logger.debug("Lowered scheduled invoker count: " + scheduledInvokers.size());
                }
                lifecycleMonitor.notifyAll();
                clearResources();
            }
            else if (isRunning()) {
                int nonPausedConsumers = getScheduledConsumerCount() - getPausedTaskCount();
                if (nonPausedConsumers < 1) {
                    logger.error("All scheduled consumers have been paused, probably due to tasks having been rejected. " +
                                 "Check your thread pool configuration! Manual recovery necessary through a start() call.");
                }
                else if (nonPausedConsumers < getConcurrentConsumers()) {
                    logger.warn("Number of scheduled consumers has dropped below concurrentConsumers limit, probably " +
                                "due to tasks having been rejected. Check your thread pool configuration! Automatic recovery " +
                                "to be triggered by remaining consumers.");
                }
            }
        }
    }
}

以上函数中主要根据变量 maxMessagesPerTask 的值来分为不同的情况处理，当然，函数中还使用了大量的代码处理异常机制的数据维护，我们主要关注程序的正常流程是如何处理的。

其实核心的处理就是调用 invokeListener() 来接收消息并**消息监听器，但是之所以两种情况分开处理，正是考虑到在无限制循环接收消息的情况下，用户可以通过设置标志位 this.running 来控制消息接收的暂停与恢复，并维护当前消息监听器的数量。

/**
 * 循环监听消息
 * @return
 * @throws JMSException
 */
private boolean executeOngoingLoop() throws JMSException {
    boolean messageReceived = false;
    boolean active = true;
    while (active) {
        synchronized (lifecycleMonitor) {
            boolean interrupted = false;
            boolean wasWaiting = false;
            // 如果当前任务已经处于**状态但是却给了暂时终止的命令
            while ((active = isActive()) && !isRunning()) {
                if (interrupted) {
                    throw new IllegalStateException("Thread was interrupted while waiting for " +
                                                    "a restart of the listener container, but container is still stopped");
                }
                if (!wasWaiting) {
                    // 如果并非处于等待状态则说明是第一次执行，需要将**任务数量减少
                    decreaseActiveInvokerCount();
                }
                // 开始进入等待状态，等待任务的恢复命令
                wasWaiting = true;
                try {
                    // 通过wait()让获取lifecycleMonitor对象锁的线程等待，等待之后获取lifecycleMonitor对象锁的线程notify或者notifyAll
                    lifecycleMonitor.wait();
                }
                catch (InterruptedException ex) {
                    // Re-interrupt current thread, to allow other threads to react.
                    // 在catch子句中，调用Thread.currentThread.interrupt()设置中断状态(因为抛出异常后中断标示会被清除)
                    // 让外界通过判断Thread.currentThread().isInterrupted()标识来决定是否终止线程还是继续下去
                    Thread.currentThread().interrupt();
                    interrupted = true;
                }
            }
            if (wasWaiting) {
                activeInvokerCount++;
            }
            if (scheduledInvokers.size() > maxConcurrentConsumers) {
                active = false;
            }
        }
        // 正常处理流程
        if (active) {
            messageReceived = (invokeListener() || messageReceived);
        }
    }
    return messageReceived;
}

如果按照正常的流程其实是不会进入while循环中的，而是直接进入函数 invokeListener() 来接收消息并**监听器，但是，我们不可能让循环一直持续下去，我们要考虑到暂停线程或者恢复线程的情况，这时，isRunning() 函数就派上用场了。

/**
 * Determine whether this container is currently running,
 * that is, whether it has been started and not stopped yet.
 * @see #start()
 * @see #stop()
 * @see #runningAllowed()
 */
@Override
public final boolean isRunning() {
   return (this.running && runningAllowed());
}

isRunning() 用来检测标志位 this.running 状态进而判断是否需要进入while 循环。由于要维护当前线程**数量，所以引入了 wasWaiting 变量，用来判断线程是否处于等待状态。如果线程首次进入等待状态，则需要调用 decreaseActiveInvokerCount() 减少线程**数量计数器。

/**
 * 减少线程**数量计数器
 */
private void decreaseActiveInvokerCount() {
   activeInvokerCount--;
   if (stopCallback != null && activeInvokerCount == 0) {
      stopCallback.run();
      stopCallback = null;
   }
}

除此以外，线程等待不是一味地采用 while 循环来控制，因为如果单纯地采用 while 循环会浪费CPU的始终周期，给资源造成巨大的浪费。这里，Spring采用的是使用全局控制变量 lifecycleMonitor 锁对象的 wait() 方法来暂停线程，所以，如果终止线程需要再次恢复的话，除了更改 this.running 标志位外，还需要调用 lifecycleMonitor.notify() 或者 lifecycle Monitor.notifyAll() 来使线程恢复。

接下来就是消息接收的处理了，调用 invokeListener() 方法。

/**
 * 消息接收的处理
 * @return
 * @throws JMSException
 */
private boolean invokeListener() throws JMSException {
   this.currentReceiveThread = Thread.currentThread();
   try {
      // 初始化资源包括首次创建的时候创建session与consumer
      initResourcesIfNecessary();
      boolean messageReceived = receiveAndExecute(this, this.session, this.consumer);
      // 改变标志位，信息成功处理
      this.lastMessageSucceeded = true;
      return messageReceived;
   }
   finally {
      this.currentReceiveThread = null;
   }
}

首先，先初始化资源，包括创建 session 与 consumer。当然初始化这两个资源我们之前有提及 cacheLevel 的事情，如果缓存了就没必要再继续创建直接获取即可。initResourcesIfNecessary() 方法的实现逻辑就是完全如之前表格所说。表面好像添加缓存没有进行那种显示的添加保存，其实这里新建的 session 与 consumer 都直接赋值给了 this 成员变量，相当于对线程来说是共享资源，所以算是缓存了。

/**
 * 初始化资源 包括首次创建的时候创建session与consumer
 * @throws JMSException
 */
private void initResourcesIfNecessary() throws JMSException {
   if (getCacheLevel() <= CACHE_CONNECTION) {
      updateRecoveryMarker();
   }
   else {
      if (this.session == null && getCacheLevel() >= CACHE_SESSION) {
         updateRecoveryMarker();
         // 创建session
         this.session = createSession(getSharedConnection());
      }
      if (this.consumer == null && getCacheLevel() >= CACHE_CONSUMER) {
         // 创建consumer
         this.consumer = createListenerConsumer(this.session);
         synchronized (lifecycleMonitor) {
            registeredWithDestination++;
         }
      }
   }
}

初始化资源完毕之后，调用 receiveAndExecute() 方法。

/**执行消息接收
 * Execute the listener for a message received from the given consumer,
 * wrapping the entire operation in an external transaction if demanded.
 * @param session the JMS Session to work on
 * @param consumer the MessageConsumer to work on
 * @return whether a message has been received
 * @throws JMSException if thrown by JMS methods
 * @see #doReceiveAndExecute
 */
protected boolean receiveAndExecute(
      Object invoker, @Nullable Session session, @Nullable MessageConsumer consumer)
      throws JMSException {

   // 用事务包裹
   if (this.transactionManager != null) {
      // Execute receive within transaction.
      TransactionStatus status = this.transactionManager.getTransaction(this.transactionDefinition);
      boolean messageReceived;
      try {
         messageReceived = doReceiveAndExecute(invoker, session, consumer, status);
      }
      catch (JMSException | RuntimeException | Error ex) {
         rollbackOnException(this.transactionManager, status, ex);
         throw ex;
      }
      this.transactionManager.commit(status);
      return messageReceived;
   }

   else {
      // Execute receive outside of transaction.
      return doReceiveAndExecute(invoker, session, consumer, null);
   }
}

doReceiveAndExecute() 包含了整个消息的接收处理过程，由于参杂着事务，所以并没有复用 JmsTemplate 模板中的方法。抛除事务代码，其核心实现还是委托 doReceiveAndExecute() 方法执行。

/**
 * Actually execute the listener for a message received from the given consumer,
 * fetching all requires resources and invoking the listener.
 * @param session the JMS Session to work on
 * @param consumer the MessageConsumer to work on
 * @param status the TransactionStatus (may be {@code null})
 * @return whether a message has been received
 * @throws JMSException if thrown by JMS methods
 * @see #doExecuteListener(javax.jms.Session, javax.jms.Message)
 */
protected boolean doReceiveAndExecute(Object invoker, @Nullable Session session,
      @Nullable MessageConsumer consumer, @Nullable TransactionStatus status) throws JMSException {

    Connection conToClose = null;
    Session sessionToClose = null;
    MessageConsumer consumerToClose = null;
    try {
        Session sessionToUse = session;
        boolean transactional = false;
        if (sessionToUse == null) {
            sessionToUse = ConnectionFactoryUtils.doGetTransactionalSession(
                obtainConnectionFactory(), this.transactionalResourceFactory, true);
            transactional = (sessionToUse != null);
        }
        if (sessionToUse == null) {
            Connection conToUse;
            if (sharedConnectionEnabled()) {
                conToUse = getSharedConnection();
            }
            else {
                conToUse = createConnection();
                conToClose = conToUse;
                conToUse.start();
            }
            sessionToUse = createSession(conToUse);
            sessionToClose = sessionToUse;
        }
        MessageConsumer consumerToUse = consumer;
        if (consumerToUse == null) {
            consumerToUse = createListenerConsumer(sessionToUse);
            consumerToClose = consumerToUse;
        }
        // 接收消息
        Message message = receiveMessage(consumerToUse);
        if (message != null) {
            if (logger.isDebugEnabled()) {
                logger.debug("Received message of type [" + message.getClass() + "] from consumer [" +
                             consumerToUse + "] of " + (transactional ? "transactional " : "") + "session [" +
                             sessionToUse + "]");
            }
            // 模板方法，当消息接收且在未处理前给子类机会做相应处理，空实现
            messageReceived(invoker, sessionToUse);
            boolean exposeResource = (!transactional && isExposeListenerSession() &&
                                      !TransactionSynchronizationManager.hasResource(obtainConnectionFactory()));
            if (exposeResource) {
                TransactionSynchronizationManager.bindResource(
                    obtainConnectionFactory(), new LocallyExposedJmsResourceHolder(sessionToUse));
            }
            try {
                // **监听器
                doExecuteListener(sessionToUse, message);
            }
            catch (Throwable ex) {
                if (status != null) {
                    if (logger.isDebugEnabled()) {
                        logger.debug("Rolling back transaction because of listener exception thrown: " + ex);
                    }
                    status.setRollbackOnly();
                }
                handleListenerException(ex);
                // Rethrow JMSException to indicate an infrastructure problem
                // that may have to trigger recovery...
                if (ex instanceof JMSException) {
                    throw (JMSException) ex;
                }
            }
            finally {
                if (exposeResource) {
                    TransactionSynchronizationManager.unbindResource(obtainConnectionFactory());
                }
            }
            // Indicate that a message has been received.
            return true;
        }
        else {
            if (logger.isTraceEnabled()) {
                logger.trace("Consumer [" + consumerToUse + "] of " + (transactional ? "transactional " : "") +
                             "session [" + sessionToUse + "] did not receive a message");
            }
            // 接收到空消息的处理
            noMessageReceived(invoker, sessionToUse);
            // Nevertheless call commit, in order to reset the transaction timeout (if any).
            if (shouldCommitAfterNoMessageReceived(sessionToUse)) {
                commitIfNecessary(sessionToUse, null);
            }
            // Indicate that no message has been received.
            return false;
        }
    }
    finally {
        JmsUtils.closeMessageConsumer(consumerToClose);
        JmsUtils.closeSession(sessionToClose);
        ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), true);
    }
}

上面函数代码看似繁杂，但是真正的逻辑并不多，大多是固定的套路，而我们最关心的就是消息接收以及监听器的**处理。

先看 receiveMessage(consumerToUse) 方法进行消息接收。

/**
 * Receive a message from the given consumer.
 * @param consumer the MessageConsumer to use
 * @return the Message, or {@code null} if none
 * @throws JMSException if thrown by JMS methods
 */
@Nullable
protected Message receiveMessage(MessageConsumer consumer) throws JMSException {
   return receiveFromConsumer(consumer, getReceiveTimeout());
}

/**
 * Actually receive a message from the given consumer.
 * @param consumer the JMS MessageConsumer to receive with
 * @param timeout the receive timeout (a negative value indicates
 * a no-wait receive; 0 indicates an indefinite wait attempt)
 * @return the JMS Message received, or {@code null} if none
 * @throws JMSException if thrown by JMS API methods
 * @since 4.3
 * @see #RECEIVE_TIMEOUT_NO_WAIT
 * @see #RECEIVE_TIMEOUT_INDEFINITE_WAIT
 */
@Nullable
protected Message receiveFromConsumer(MessageConsumer consumer, long timeout) throws JMSException {
    if (timeout > 0) {
        return consumer.receive(timeout);
    }
    else if (timeout < 0) {
        return consumer.receiveNoWait();
    }
    else {
        return consumer.receive();
    }
}

我们终于看到了我们认识的代码了，层层调用就是为了找到这段代码。接收到消息后，监听器容器将消息 Message 对象传递给我们的自定义监听器，我们可以在监听器的 onMessage() 方法中对接收到的消息做自定义逻辑处理。

/**
 * **自定义监听器，将接收到的消息传给自定义监听器
 * Execute the specified listener,
 * committing or rolling back the transaction afterwards (if necessary).
 * @param session the JMS Session to operate on
 * @param message the received JMS Message
 * @throws JMSException if thrown by JMS API methods
 * @see #invokeListener
 * @see #commitIfNecessary
 * @see #rollbackOnExceptionIfNecessary
 * @see #convertJmsAccessException
 */
protected void doExecuteListener(Session session, Message message) throws JMSException {
    if (!isAcceptMessagesWhileStopping() && !isRunning()) {
        if (logger.isWarnEnabled()) {
            logger.warn("Rejecting received message because of the listener container " +
                        "having been stopped in the meantime: " + message);
        }
        rollbackIfNecessary(session);
        throw new MessageRejectedWhileStoppingException();
    }

    try {
        invokeListener(session, message);
    }
    catch (JMSException | RuntimeException | Error ex) {
        rollbackOnExceptionIfNecessary(session, ex);
        throw ex;
    }
    commitIfNecessary(session, message);
}

/**
 * Invoke the specified listener: either as standard JMS MessageListener
 * or (preferably) as Spring SessionAwareMessageListener.
 * @param session the JMS Session to operate on
 * @param message the received JMS Message
 * @throws JMSException if thrown by JMS API methods
 * @see #setMessageListener
 */
@SuppressWarnings("rawtypes")
protected void invokeListener(Session session, Message message) throws JMSException {
    Object listener = getMessageListener();

    if (listener instanceof SessionAwareMessageListener) {
        doInvokeListener((SessionAwareMessageListener) listener, session, message);
    }
    else if (listener instanceof MessageListener) {
        doInvokeListener((MessageListener) listener, message);
    }
    else if (listener != null) {
        throw new IllegalArgumentException(
            "Only MessageListener and SessionAwareMessageListener supported: " + listener);
    }
    else {
        throw new IllegalStateException("No message listener specified - see property 'messageListener'");
    }
}

/**
 * Invoke the specified listener as standard JMS MessageListener.
 * <p>Default implementation performs a plain invocation of the
 * {@code onMessage} method.
 * @param listener the JMS MessageListener to invoke
 * @param message the received JMS Message
 * @throws JMSException if thrown by JMS API methods
 * @see javax.jms.MessageListener#onMessage
 */
protected void doInvokeListener(MessageListener listener, Message message) throws JMSException {
    listener.onMessage(message);
}

通过层层调用，最终提取监听器并使用 listener.onMessage(message) **了监听器，也就是**了用户自定义的监听器逻辑。

doReceiveAndExecute() 方法中还有一句重要的代码很容易被忽略掉，commitIfNecessary(sessionToUse, null) 完成的功能是 session.commit()。

完成消息服务的事务提交，涉及两个事务。我们常说的 DefaultMessageListenerContainer 增加了Spring的事务支持，是通用的事务，也就是说我们在消息接收过程中如果产生其他操作，比如向数据库中插入数据，一旦出现异常时就需要全部回滚，包括回滚插入数据库中的数据，也就是外层大事务。但是，除了我们常说的事务之外，对于消息本身还有一个事务，当接收一个消息的时候，必须使用事务提交的方式，这是在告诉消息服务器本地已经正常接收消息，消息服务器接收到本地的事务提交后便可以将此消息删除，否则，当前消息会被其他接收者重新接收。

IV. 总结

Spring整合 JMS 消息队列的实现思路和 JDBC 非常相似，都应用了模板设计模式，同时Spring的事务不仅仅只是用在数据库之上，对于消息队列也同样适用。

除此以外，对于消息队列的接收，Spring对于消息接收提供了监听器能够一直监听是否有消息的发送。对于消息监听，其实相当于底层实现是Spring开辟子线程循环监听是否有消息能够接收，如果有消息则接收传给我们自定义的消息监听器进行逻辑处理。

Spring源码——消息队列

前言

I. 单独使用ActiveMQ

安装ActiveMQ

JMS独立使用

① 发送端

② 接收端

③ 测试结果

分析

II. Spring整合ActiveMQ

添加依赖

配置文件

发送端

接收端

测试结果

监听器

① 创建消息监听器

② 修改配置文件

③ 测试结果

III. 源码分析

JmsTemplate

① 通用代码抽取

② 发送消息的实现

③ 接收消息的实现

消息监听器

① 初始化容器

② 启动消息监听任务

③ 监听任务实现细节

IV. 总结

参考阅读

相关推荐