探究SpringJMS+ActiveMQ消息阻塞之谜

作者: coder_jerry | 来源:发表于2017-09-22 19:20 被阅读279次

0x00 背景介绍

最近遇到了一个消息队列阻塞的问题，查了好几天，才终于把这个问题解决。
首先交代下我们的环境：支付业务，使用ActiveMQ配合Spring-jms收发消息，Spring版本3.2.16.RELEASE，ActiveMQ客户端版本5.11.1，ActiveMQ服务端版本5.13.2。消息队列消费者使用Spring-jms的DefaultMessageListenerContainer作为消息监听容器，用org.springframework.jms.connection.CachingConnectionFactory对org.apache.activemq.ActiveMQConnectionFactory进行了封装，提供缓存功能。DefaultMessageListenerContainer的并发配置为1-10。
在较长的时间里，系统运行总体上较为稳定，但是期间发生过多次消息阻塞。之前查到过由于端口占用问题导致消息阻塞（https://issues.apache.org/jira/browse/AMQ-6362），只能通过重启解决，后来通过修改ActiveMQ client源码请求超时时间解决了这一问题（此问题在ActiveMQ 5.14.5得到修复）。但是还有一种偶发的消息阻塞，时不常就会来那么一次，让人防不胜防。因为这个问题频发，之前还用Python写了一套监控MQ队列消息阻塞情况的代码，通过微信企业号接口发送。所以能大概知道每次阻塞的持续时间。这种阻塞发生概率较低，而且每次可能阻塞的时间不定，但大部分在30~40分钟左右。
另外，我使用了两个DefaultMessageListenerContainer实例，分别对应两个队列消费者。除业务处理监听外，其他配置都一样，但是每次阻塞都发生在其中一个固定队列。这种情况一度让我很困惑。直到最近再进行代码review，跟踪spring和activemq的源代码时，才发现一点头绪。

0x01 相关概念简介

先简单介绍下几个涉及的概念：
DefaultMessageListenerContainer （后文简称DMLC）：SpringJMS中用于异步消息监听的管理类。主要原理简单介绍下。首先通过内部初始化建立一个taskExecutor（默认实现是SimpleAsyncTaskExecutor），用于执行消息监听任务，任务通过DMLC内部类AsyncMessageListenerInvoker表示。默认情况下，SimpleAsyncTaskExecutor每次都会开启一个新线程。AsyncMessageListenerInvoker是实现了Runnable接口的任务抽象，在其executeOngoingLoop方法中不断调用MessageConsumer#receive(this.receiveTimeout)收取消息（见DMLC父类AbstractPollingMessageListenerContainer代码），这里默认的receiveTimeout是1秒。在接收到消息后，将会根据选定DMLC属性maxConcurrentConsumers（jms:listener-container标签中concurrency的最大值）和当前空闲消费线程数量，对消费线程数量，也就是同时运行的AsyncMessageListenerInvoker进行扩容，即由taskExecutor新建线程（通过DefaultMessageListenerContainer调用持有的ConnectionFactory方法创建session和consumer的方法获取MessageConsumer对象）接收消息。每个AsyncMessageListenerInvoker一次最大执行的任务数量，即MessageConsumer#receive(this.receiveTimeout)方法调用次数可以通过DMLC的maxMessagesPerTask属性制定，默认为Integer.MIN_VALUE，当指定为大于0的值时，在执行完maxMessagesPerTask次消息收取后，将会销毁当前线程。
SingleConnectionFactory（后文简称SCF）：一个JMS ConnectionFactory的适配器，通过JDK Proxy代理Connection（参见SingleConnectionFactory$SharedConnectionInvocationHandler），总是获取同一个Connection，且忽略close。与消息监听容器配合使用的作用是，在多个容器间共享一个JMS Connection。
CachingConnectionFactory （后文简称CCF）：在SingleConnectionFactory的基础上，添加了Session、MessageConsumer、MessageProducer的缓存功能，与SingleConnectionFactory类似，使用JDK Proxy代理了Session，具体可以查看CachingConnectionFactory$CachedSessionInvocationHandler这个内部类，它包装了原生的Session，主要提供MessageConsumer和MessageProducer的缓存功能。缓存session数量由参数sessionCacheSize决定，默认为1。

0x02 探究MQ阻塞之谜

通过阅读源码和javadoc了解到，在DMLC的注释文档中，写明了DefaultMessageListenerContainer不能与CachingConnectionFactory配合进行动态扩容。这是因为CachingConnectionFactory更倾向于让监听容器自己处理缓存。在同时停止和重启一个监听容器时，只会在独立的且被容器本身缓存的连接上起作用，而不会在外部缓存的连接上起作用。

Note: Don't use Spring's org.springframework.jms.connection.CachingConnectionFactory in combination with dynamic scaling. Ideally, don't use it with a message listener container at all, since it is generally preferable to let the listener container itself handle appropriate caching within its lifecycle. Also, stopping and restarting a listener container will only work with an independent, locally cached Connection - not with an externally cached one.

在StackOverflow中Spring-jms的开发主管也说明了为什么DMLC不能使用CCF，简而言之，CCF由于缓存了Session和MessageConsumer，在DMLC对消费线程进行缩容时，虽然消费线程销毁了，但是消费者在，Broker还会一直给MessageConsumer发消息。
其实查到这，就完全够用了，把CCF换掉，程序应该就可以跑了。但那显然不符合咱们对技术刨根问题的风格，于是咱们还是探究一下MQ为什么会阻塞的实际原因。现在虽然是知道DMLC不能和CCF一起使用，但因为我们的DMLC并未指定maxMessagesPerTask参数，并不存在缩容的情况，所以实际原因还需要深入调查。
可能有小伙伴还不太了解ActiveMQ客户端的消费逻辑，这里我们插播下ActiveMQ消费逻辑。

0x02.1 ActiveMQ消费逻辑

ActiveMQ在消费消息时，并非直接从Broker进行网络通信，取消息进行消费。我们通过FailoverTransport+TcpTransport举例，也就是failover:(tcp:...)形式的brokerUrl。

activemq-client的消费过程大致可以总结为，一个线程和broker进行通信，取得消息，放到一个缓存中，而另一个线程则不断地尝试从这个缓存中获取消息。
为了说明ActiveMQ的消费逻辑，我从一个完整的线程堆栈中，拉出了以下两条线程堆栈。

负责和broker通信（tcp长连接）取消息的线程：

"ActiveMQ Transport: tcp://jms.xxx/192.168.30.xx:61618@54095" prio=10 tid=0x00007f2bf0001000 nid=0x5f94 runnable [0x00007f2e470ef000]
   java.lang.Thread.State: RUNNABLE
    at java.net.SocketInputStream.socketRead0(Native Method)
    at java.net.SocketInputStream.read(SocketInputStream.java:152)
    at java.net.SocketInputStream.read(SocketInputStream.java:122)
    at org.apache.activemq.transport.tcp.TcpBufferedInputStream.fill(TcpBufferedInputStream.java:50)
    at org.apache.activemq.transport.tcp.TcpTransport$2.fill(TcpTransport.java:609)
    at org.apache.activemq.transport.tcp.TcpBufferedInputStream.read(TcpBufferedInputStream.java:58)
    at org.apache.activemq.transport.tcp.TcpTransport$2.read(TcpTransport.java:594)
    at java.io.DataInputStream.readInt(DataInputStream.java:387)
    at org.apache.activemq.openwire.OpenWireFormat.unmarshal(OpenWireFormat.java:258)
    at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:221)
    at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:213)
    at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:196)
    at java.lang.Thread.run(Thread.java:745)

使用spring-jms DMLC消费消息的线程：

"org.springframework.jms.listener.DefaultMessageListenerContainer#0-4" prio=10 tid=0x00007f2b4c076800 nid=0xb442 in Object.wait() [0x00007f2e3f3f2000]
   java.lang.Thread.State: TIMED_WAITING (on object monitor)
    at java.lang.Object.wait(Native Method)
    - waiting on <0x00000007ff81a8e8> (a java.lang.Object)
    at org.apache.activemq.**FifoMessageDispatchChannel**.dequeue(FifoMessageDispatchChannel.java:74)
    - locked <0x00000007ff81a8e8> (a java.lang.Object)
    at org.apache.activemq.ActiveMQMessageConsumer.dequeue(ActiveMQMessageConsumer.java:482)
    at org.apache.activemq.ActiveMQMessageConsumer.receive(ActiveMQMessageConsumer.java:635)
    at org.springframework.jms.connection.CachedMessageConsumer.receive(CachedMessageConsumer.java:74)
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveMessage(AbstractPollingMessageListenerContainer.java:430)
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:310)
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:243)
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1103)
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1095)
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:992)
    at java.lang.Thread.run(Thread.java:745)

这里，这个org.apache.activemq.FifoMessageDispatchChannel（由ActiveMQMessageConsumer持有，属性名称unconsumedMessages）就是用于缓存消息的类，其内部持有一个LinkedList<MessageDispatch>，这个MessageDispatch就是从broker分发过来还未消费的消息封装。

0x02.2 基于源码，大胆猜测

拉回整体，继续探究MQ阻塞之谜。
既然MQ的消费分两步，那应该就是在MessageConsumer收取到消息到本地后，消费线程没有及时调用MessageConsumer#receive()方法，导致消息一直在本地积压。但是broker不知道啊，只要这个MessageConsumer还在，它就会一直给它发消息。除非它不在了，才会把已经分发但未消费的消息分配给其他的MessageConsumer。
我按照线上环境，在本机使用单元测试进行模拟。使用反射获取ActiveMQMessageConsumer的消息缓存unconsumedMessages，每5秒打印一次所有阻塞的消息（包括consumerId, sessionId）。并在log4j2配置中，将org.springframework.jms和org.apache.activemq的打印级别调整为DEBUG。
果然发生了阻塞，通过在所有日志中查找阻塞消息的consumerId，（样例：session: ActiveMQSession {id=ID:mypc-58436-1505458977674-3:1:2,started=true}, consumer:ActiveMQMessageConsumer { value=ID:mypc-58436-1505458977674-3:1:2:1, started=true },）可以发现，在15:03:06到15:05:04之间，没有任何线程尝试从该阻塞消息对应的session。

image.png

修改DMLC的并发配置（listener-container的concurrency），从1-10变为固定值10，复现，说明和DMLC的动态改变消费线程无关。
问题查到这里，能断定的是，本文MQ的阻塞肯定和消费线程没有调用MessageConsumer#receive方法有关。但究竟为什么，是消费线程获取不到Session，还是获取不到MessageConsumer，还是获取到了MessageConsumer不能调用receive方法？还不得而知。

顺带说一下ActiveMQSession的toString()格式：

"ActiveMQSession {id=" + connectionId + ":" + 全局唯一递增序列号 + ",started=" + started.get() + "}" // 参见 ActiveMQSession,ActiveMQConnection,LongSequenceGenerator,SessionId代码

对于zw_123-41988-1505188712169-1:1:13613，zw_123-41988-1505188712169-1:1是connectionId，而13613是指自JVM启动起来创建的ActiveMQSession序列号。

看了半天DMLC了，既然没有什么线索了，就暂时看下其他的。
CachingConnectionFactory中用于缓存Session的cachedSessions属性是一个Map<Integer, LinkedList<Session>>，根据session类型（0-SESSION_TRANSACTED，1-AUTO_ACKNOWLEDGE，2-CLIENT_ACKNOWLEDGE，3-DUPS_OK_ACKNOWLEDGE）作为key获取一个Session List。这个sessionList的引用在创建Session代理（CachingConnectionFactory.CachedSessionInvocationHandler）时被传递进去，并作为实例变量sessionList引用，默认情况下，这个sessionList的size为1。而在Session代理实现CachingConnectionFactory.CachedSessionInvocationHandler中的cachedConsumers和cachedProducers负责缓存消费者和生产者。

以下为CachingConnectionFactory中获取Session和返还Session涉及Session缓存（sessionList）的处理代码。

// CachingConnectionFactory.java
/**
* Checks for a cached Session for the given mode.
*/
protected Session getSession(Connection con, Integer mode) throws JMSException {
LinkedList<Session> sessionList;
synchronized (this.cachedSessions) {
  sessionList = this.cachedSessions.get(mode); // 获取缓存session列表
  if (sessionList == null) {
  sessionList = new LinkedList<Session>();
  this.cachedSessions.put(mode, sessionList);
  }
}
Session session = null;
synchronized (sessionList) {
  if (!sessionList.isEmpty()) {
  session = sessionList.removeFirst(); // 如果不为空，就取第一个session。默认情况下，sessionCacheSize数量为1，在cachedSessions没有缓存session时，会创建session
  }
}
if (session != null) { // 从缓存中取得session
  if (logger.isTraceEnabled()) {
  logger.trace("Found cached JMS Session for mode " + mode + ": " +
    (session instanceof SessionProxy ? ((SessionProxy) session).getTargetSession() : session));
  }
}
else { // 如果没有缓存session，就创建session
  Session targetSession = createSession(con, mode); // con就是CachingConnectionFactory中的target
  if (logger.isDebugEnabled()) {
  logger.debug("Creating cached JMS Session for mode " + mode + ": " + targetSession);
  }
  session = getCachedSessionProxy(targetSession, sessionList); // 使用jdk方式代理session
}
return session;
}

// CachingConnectionFactory内部类 CachedSessionInvocationHandler，实现了InvocationHandler接口，负责Session代理
private class CachedSessionInvocationHandler implements InvocationHandler {
// ...
private void logicalClose(Session proxy) throws JMSException { // 逻辑关闭，没有实际关闭session，而是放到sessionList缓存
// Preserve rollback-on-close semantics.
if (this.transactionOpen && this.target.getTransacted()) {
  this.transactionOpen = false;
  this.target.rollback();
}
// Physically close durable subscribers at time of Session close call.
for (Iterator<Map.Entry<ConsumerCacheKey, MessageConsumer>> it = this.cachedConsumers.entrySet().iterator(); it.hasNext();) {
  Map.Entry<ConsumerCacheKey, MessageConsumer> entry = it.next();
if (entry.getKey().subscription != null) {
  entry.getValue().close();
  it.remove();
}
}
// Allow for multiple close calls...
boolean returned = false;
synchronized (this.sessionList) {
  if (!this.sessionList.contains(proxy)) {
  this.sessionList.addLast(proxy);
  returned = true;
  }
}
if (returned && logger.isTraceEnabled()) {
  logger.trace("Returned cached Session: " + this.target);
}
}

// ...
}

DMLC每次想要获得消息，就先得获取Session。在使用CCF时，CCF会首先判断Session缓存里有没有可用的Session，如果有就取出直接使用，如果没有就创建一个新的Session，并在Session使用完毕后尝试将Session放到Session缓存中，如果Session缓存已经满了，就执行Session的物理关闭。

0x02.3 通过Eclipse MAT查看对象信息

基本逻辑差不多都了解了，那么我们查看下在阻塞发生时，Java堆中阻塞消息相关的MessageConsumer、Session信息。（啥？你还打了消息阻塞时候的堆信息？--对呀~）
使用Eclipse MAT打开Java堆信息，使用OQL查询消息阻塞的相关Session和Consumer：

SELECT c, c.session, c.unconsumedMessages.list.size FROM org.apache.activemq.ActiveMQMessageConsumer c WHERE (c.unconsumedMessages.list.size > 0)

查询到ActiveMQMessageConsumer和ActiveMQSession后，进行查看。在查看相关的ActiveMQSession时，发现了问题。

image.png

如图中所示，消息阻塞的MessageConsumer相关的Session有两个MessageConsumer。项目中使用两个DMLC分别监听两个队列，使用同一个CCF封装ActiveMQConnectionFactory获取连接和Session。这两个消费者很显然就是由两个DMLC调用同一个Session创建的。那么，是不是消费线程虽然获取了Session，但一直在处理另一个MessageConsumer，而忽略了被阻塞消息所在的MessageConsumer呢？？？

0x02.4 锁定问题-还是日志大法好

有了这个设想，就有了一点方向，希望能从日志中找到些蛛丝马迹。
正巧近期又有次阻塞，直接从线上查日志。功夫不负有心人，通过查找消息阻塞相关的session {id=ID:zw_123-41988-1505188712169-1:1:13613,started=true}，从下面这段看似平淡无奇的日志里，终于看到了一丝曙光。

1203078 2017-09-18 06:09:53,996|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Creating new transaction with name [org.springframework.jms.listener.DefaultMessageListenerContainer#0]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
1203079 2017-09-18 06:09:53,996|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Creating cached JMS Session for mode 0: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13613,started=true}
1203080 2017-09-18 06:09:53,996|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Created JMS transaction on Session [Cached JMS Session: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13613,started=true}] from Connection [Shared JMS Connection: ActiveMQConnection {id=ID:zw_123-41988-1505188712169-1:1,clientId=ID:zw_123-41988-1505188712169-0:1,started=true}]
1203081 2017-09-18 06:09:53,998|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Creating cached JMS MessageConsumer for destination [queue://ConsumptionOrder.Has.Paied]: ActiveMQMessageConsumer { value=ID:zw_123-41988-1505188712169-1:1:13613:1, started=true }
... 省略...
1292825 2017-09-18 06:37:29,672|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Initiating transaction commit
1292826 2017-09-18 06:37:29,672|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Committing JMS transaction on Session [Cached JMS Session: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13613,started=true}]
1292827 2017-09-18 06:37:29,672|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|ID:zw_123-41988-1505188712169-1:1:13613 Transaction Commit :null
1292828 2017-09-18 06:37:29,672|TRACE|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Returned cached Session: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13613,started=true}
1292829 2017-09-18 06:37:29,672|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|Creating new transaction with name [org.springframework.jms.listener.DefaultMessageListenerContainer#1]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
1292830 2017-09-18 06:37:29,672|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Creating new transaction with name [org.springframework.jms.listener.DefaultMessageListenerContainer#0]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
1292831 2017-09-18 06:37:29,672|TRACE|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|**Found cached JMS Session for mode 0: ActiveMQSession** {id=ID:zw_123-41988-1505188712169-1:1:13613,started=true}
1292832 2017-09-18 06:37:29,672|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|Created JMS transaction on Session [Cached JMS Session: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13613,started=true}] from Connection [Shared JMS Connection: ActiveMQConnection {id=ID:zw_123-41988-1505188712169-1:1,clientId=ID:zw_123-41988-1505188712169-0:1,started=true}]
1292833 2017-09-18 06:37:29,673|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Creating cached JMS Session for mode 0: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13629,started=true}
1292834 2017-09-18 06:37:29,673|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Created JMS transaction on Session [Cached JMS Session: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13629,started=true}] from Connection [Shared JMS Connection: ActiveMQConnection {id=ID:zw_123-41988-1505188712169-1:1,clientId=ID:zw_123-41988-1505188712169-0:1,started=true}]
1292835 2017-09-18 06:37:29,674|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|Creating cached JMS MessageConsumer for destination [**queue://queue.B**]: ActiveMQMessageConsumer { value=ID:zw_123-41988-1505188712169-1:1:13613:2, started=true }
1292836 2017-09-18 06:37:29,675|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Creating cached JMS MessageConsumer for destination [**queue://queue.A**]: ActiveMQMessageConsumer { value=ID:zw_123-41988-1505188712169-1:1:13629:1, started=true }

对看似平淡无奇的日志的分析：
DefaultMessageListenerContainer#0 这个DMLC实例负责处理队列A，DefaultMessageListenerContainer#1这个DMLC实例负责处理队列B。每个DMLC实例在处理消息时，会根据配置可伸缩地使用1或多条线程去处理任务（伸缩逻辑见DMLC，处理消息逻辑见其内部类AsyncMessageListenerInvoker）。此时有两个线程#0-3负责处理队列A，#1-1负责处理队列B。但是由于使用了CachingConnectionFactory（CachingConnectionFactory实例只有一个，其实例变量cachedSessions负责保存创建的session，默认大小为1），#0-3线程在结束本次Session使用后，将Session放回了缓存中（1292828行）。此时恰逢#1-1线程请求Session，一看缓存中有可用Session，便取出使用（1292831行）。这个Session并非原生Session，而是CachingConnectionFactory通过JDK Proxy进行代理创建的Session代理（逻辑见内部类CachedSessionInvocationHandler），在Session代理中，对MessageConsumer进行了缓存。但是由于#1-1线程要处理的是队列B，不能使用#0-3之前创建的处理队列A的MessageConsumer，所以就创建了一个新的MessageConsumer负责处理队列B（1292835行）。
此时，线程#1-1将调用新创建的MessageConsumer的receive方法接收消息。在spring-jms中DMLC中规定了receive的默认超时时间为1秒（见DMLC父类AbstractPollingMessageListenerContainer.DEFAULT_RECEIVE_TIMEOUT），如果在1秒内没有接收到消息，将返回Session到缓存，并继续下一个循环。
注意，在接收消息的这一次循环中，相关的这个Session已经被该线程霸占，其他线程不能获取该Session，也就是说，该Session的另一个MessageConsumer中接收到的队列A的消息将得不到处理！
然而正常情况下，这种影响较小。在以下情况下才有可能出现问题：

在一次消息接收循环的末尾，线程将Session返回到缓存池（CachingConnectionFactory中的cachedSessions）时，缓存池有空闲空间。因为如果缓存空间满了，程序将对Session做物理关闭，Session创建的MessageConsumer和MessageProducer将一并关闭，可能存在的未消费消息也会得到释放。
在1成功的条件下，该线程在下一个消息接收循环的开始，又一次成功获取该Session。因为一旦其他线程获取该Session，就将调用另一个队列进行消费。这一步也是一个概率问题，虽然从缓存池取出操作使用removeFirst，返还操作使用addLast，即取、还操作分别操作首尾，但是由于sessionCacheSize默认设置为1，由于Session的返还和取出操作极快，在并发不严重的情况下，往往就会取到同一个Session。
在1、2循环发生时，线程将霸占该Session，即在该Session的处理队列A的MessageConsumer被“遗忘”时，ActiveMQ又将消息分配给它处理。

另，本项目现有配置为：DefaultMessageListenerContainer配合CachingConnectionFactory使用，使用2个DefaultMessageListenerContainer实例分别监听2个队列，并发数为1-10，其connectionFactory为一个CachingConnectionFatory，CachingConnectionFactory的sessionCacheSize为1。推测在使用DefaultMessageListenerContainer配合CachingConnectionFactory使用时，使用多个DefaultMessageListenerContainer共享一个CachingConnectionFactory时，有可能会出现该问题。
继续上面的条件分析，1、2看起来好像很难发生。但设想，如果所有的线程总是以一定的速率操作从Session缓存池取出、返还Session，那么是不是同一个线程拿到相同Session的概率就更大呢？那这个一定速率又是怎么来的呢，是不是总是收不到消息，大家都是按照默认超时时间1秒来循环操作缓存池呢？

回顾线上环境当时的情况，消息使用量确实不高。因此消息阻塞的情况一直持续了大概40分钟，才终于被打破。原因是条件1中所说的“缓存池为空”条件不成立了，线程#1-1在返还Session，碰巧另一个线程同时放了个Session进去（又是老冤家#0-3, - -!），线程#1-1发现缓存池满了（其实就是默认值1），决定对Session进行物理关闭。Session的关闭才终于结束了其创建的MessageConsumer的生命，而MessageConsumer也终于肯放弃对未消费消息的持有，依依不舍地去了（参见ActiveMQMessageConsumer#close()）。。。之后ActiveMQ客户端发送Session和Consumer、Producer的删除信息给ActiveMQ Broker，未消费的MessageDispatch也将被重新分配给其他的Consumer进行消费。
以下是阻塞情况被打破的日志。查看第3、4行，#0-3线程返还Session到Session缓存池，近乎同时，#1-1发现Session缓存满了，只能把Session物理关闭。此时其关联的MessageConsumer也被关闭，至此未消费的消息才被回收重新分发。

2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|ID:zw_123-41988-1505188712169-1:1:13686 Transaction Commit :null
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|ID:zw_123-41988-1505188712169-1:1:13613 Transaction Commit :null
**2017-09-18 07:19:49,910|TRACE|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Returned cached Session: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13686,started=true}
**
**2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|Closing cached Session: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13613,started=true}
**
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Creating new transaction with name [org.springframework.jms.listener.DefaultMessageListenerContainer#0]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
2017-09-18 07:19:49,910|TRACE|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Found cached JMS Session for mode 0: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13686,started=true}
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|remove: ID:zw_123-41988-1505188712169-1:1:13613:2, lastDeliveredSequenceId:-1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Created JMS transaction on Session [Cached JMS Session: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13686,started=true}] from Connection [Shared JMS Connection: ActiveMQConnection {id=ID:zw_123-41988-1505188712169-1:1,clientId=ID:zw_123-41988-1505188712169-0:1,started=true}]
2017-09-18 07:19:49,910|TRACE|org.springframework.jms.listener.DefaultMessageListenerContainer#0-3|Found cached JMS MessageConsumer for destination [queue://queue.A]: ActiveMQMessageConsumer { value=ID:zw_123-41988-1505188712169-1:1:13686:1, started=true }
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:app_456-51856-1505195602738-5:9324:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:app_456-30343-1505195633167-5:9537:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:zw_123-45783-1505195534974-5:9184:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:app_456-30343-1505195633167-5:9539:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:zw_123-45783-1505195534974-5:9188:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:zw_123-63173-1505195538255-1:9237:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:app_456-30343-1505195633167-5:9542:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:app_456-51856-1505195602738-5:9331:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:zw_123-45783-1505195534974-5:9192:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:zw_123-45783-1505195534974-5:9194:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:app_789-25467-1501664276715-5:50772:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:app_456-51856-1505195602738-5:9337:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:zw_123-63173-1505195538255-1:9248:1:1:2
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:app_456-51856-1505195602738-5:9340:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:zw_123-63173-1505195538255-1:9252:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:app_789-36182-1501664210853-5:50813:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:app_456-30343-1505195633167-5:9555:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:app_789-33412-1505357006369-3:316:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:app_789-36182-1501664210853-5:50816:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|on close, rollback duplicate: ID:zw_123-45783-1505195534974-5:9204:1:1:1
2017-09-18 07:19:49,910|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|remove: ID:zw_123-41988-1505188712169-1:1:13613:1, lastDeliveredSequenceId:90527387
2017-09-18 07:19:49,911|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|Creating new transaction with name [org.springframework.jms.listener.DefaultMessageListenerContainer#1]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
2017-09-18 07:19:49,911|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|Creating cached JMS Session for mode 0: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13687,started=true}
2017-09-18 07:19:49,911|DEBUG|org.springframework.jms.listener.DefaultMessageListenerContainer#1-1|Created JMS transaction on Session [Cached JMS Session: ActiveMQSession {id=ID:zw_123-41988-1505188712169-1:1:13687,started=true}] from Connection [Shared JMS Connection: ActiveMQConnection {id=ID:zw_123-41988-1505188712169-1:1,clientId=ID:zw_123-41988-1505188712169-0:1,started=true}]

0x03 问题解决

原因分析的差不多了，解决起来也比较简单。只要破坏掉以上几个条件之一即可。但为了避免引入其他问题，还是依Spring-jms所言，换掉CachingConnectionFactory，使用其他的ConnectionFactory。
在选择ConnectionFactory时，也遇到了选择障碍。是直接使用ActiveMQConnectionFactory，还是使用Spring-jms的SCF，还是使用PooledConnectionFactory呢？

0x03.1 PooledConnectionFactory为啥不缓存MessageConsumer？

提到了PooledConnectionFactory，就想为什么PooledConnectionFactory不对Consumer进行缓存呢？看了看PooledConnectionFactory的源码，才发现原来注释就说的很明白。
org.apache.activemq.jms.pool.PooledConnectionFactory注释：

A JMS provider which pools Connection, Session and MessageProducer instances so it can be used with tools like Camel and Spring's JmsTemplate and MessagListenerContainer. Connections, sessions and producers are returned to a pool after use so that they can be reused later without having to undergo the cost of creating them again. b>NOTE: while this implementation does allow the creation of a collection of active consumers, it does not 'pool' consumers. Pooling makes sense for connections, sessions and producers, which are expensive to create and can remain idle a minimal cost. Consumers, on the other hand, are usually just created at startup and left active, handling incoming messages as they come. When a consumer is complete, it is best to close it rather than return it to a pool for later reuse: this is because, even if a consumer is idle, ActiveMQ will keep delivering messages to the consumer's prefetch buffer, where they'll get held until the consumer is active again. If you are creating a collection of consumers (for example, for multi-threaded message consumption), you might want to consider using a lower prefetch value for each consumer (e.g. 10 or 20), to ensure that all messages don't end up going to just one of the consumers. See this FAQ entry for more detail: http://activemq.apache.org/i-do-not-receive-messages-in-my-second-consumer.html Optionally, one may configure the pool to examine and possibly evict objects as they sit idle in the pool. This is performed by an "idle object eviction" thread, which runs asynchronously. Caution should be used when configuring this optional feature. Eviction runs contend with client threads for access to objects in the pool, so if they run too frequently performance issues may result. The idle object eviction thread may be configured using the setTimeBetweenExpirationCheckMillis method. By default the value is -1 which means no eviction thread will be run. Set to a non-negative value to configure the idle eviction thread to run.

0x03.2 简单的性能比对

但是如果不缓存消费者，效率真是比较低，还是选择缓存消费者吧。
排除了PooledConnectionFactory后，我选择了以下维度进行测试。
不同的DMLC cacheLevel：CACHE_NONE, CACHE_CONNECTION, CACHE_SESSION, CACHE_CONSUMER, AUTO
不同的DMLC connectionFactory： ActiveMQConnectionFactory, SingleConnectionFactory
不同的DMLC注入方式：使用jms:listener-container标签, 使用bean标签注册DMLC实例

经过一番非常不严谨的比对后，发现性能最高的组合是（基于JDK1.7）：
DMLC注入方式（使用bean标签注册DMLC实例）+connectionFactory（SingleConnectionFactory ）+cacheLevel（CACHE_CONSUMER）。

0x03.03 使用Wireshark和Python Pandas探究性能谜题

我一直以为使用jms:listener-container标签选择default也就是使用DMLC的方式和使用bean注入DMLC的方式效率一样，可实际测试才发现，使用bean的方式要比使用listener-container快1~3倍（注意是不严谨的测试哦，你测出来可能和我不一样的）。
百思不得其解之下，我尝试从它们的流量包中分析。
借助Wireshark，我分别抓了两种方式下的流量数据，每次都是收发1000条消息。dmlc-tag表示使用jms标签注入DMLC，dmlc-bean表示使用bean标签注入DMLC（名字不好没关系，只是代号而已--）。因为测试的量比较少，两种方式tag和bean分别使用5194ms和3246ms就完成了。
再看流量包的分析（WireShark 统计-IPv4 Statistics-All Addresses），输入显示过滤器“tcp.port == 61618”，发现使用bean方式的网络流量包相较使用jms tag的减少了3000+。

image.png

可是这样看，非常不直观，是哪种包少了呢？
为了更直观的查看，使用tshark将流量数据包导出成csv进行分析。

tshark.exe -r dmlc-tag.pcap -T fields -e frame.number -e frame.time_epoch -e ip.src -e ip.dst -e ip.proto -e openwire -E header=y -E separator="-" -E quote=n  openwire > dmlc-tag.csv
tshark.exe -r dmlc-bean.pcap -T fields -e frame.number -e frame.time_epoch -e ip.src -e ip.dst -e ip.proto -e openwire -E header=y -E separator="-" -E quote=n  openwire > dmlc-bean.csv

再借助Python Pandas进行粗略的分析：

import pandas as pd
tagdf = pd.read_csv('920pcap\dmlc-tag.csv',delimiter='-')
beandf = pd.read_csv('920pcap\dmlc-bean.csv',delimiter='-')

tagdf['command'] = tagdf['openwire'].str.split(',').str.get(0)
beandf['command'] = beandf['openwire'].str.split(',').str.get(0)

tagdf.groupby('command')['command'].count()
beandf.groupby('command')['command'].count()

image.png

对两组数据包含的openwire协议命令求差集
set(tagdf['command'].unique())^set(beandf['command'].unique())

最终发现，bean的response要少(1010 vs. 44)，而且没有TransactionInfo。

又经过一番源码的探索，发现，如果DMLC的sessionTransacted属性没有指明true，则将不使用事务。

在Spring-jms源码org.springframework.jms.config.JmsListenerContainerParser中可以发现，acknowledge可以影响ListenerContainer的两个属性sessionTransacted和sessionAcknowledgeMode。

image.png

所以从jms:listener-container标签方式切换到bean注入方式，使用事务需要注意sessionTransacted这个属性，光是设置transactionManager可能不行。

下面给出使用jms:listener-container和bean方式注入DMLC的代码，希望小伙伴们避免犯我这样的错误。

<jms:listener-container
        connection-factory="consumerJmsConnectionFactory"
        transaction-manager="consumerJmsTransactionManager"
        concurrency="1-10"
        **acknowledge="transacted"**
        cache="consumer"
    >
        <jms:listener destination="dest.queue" ref="testConsumer" method="receive"/> 
    </jms:listener-container>

<bean id="consumptionOrderListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<property name="delegate" ref="testConsumer"/>
<property name="defaultListenerMethod" value="receive"/>
<!--<property name="messageConverter"-->
</bean>
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="consumerJmsConnectionFactory"/>
<property name="transactionManager" ref="consumerJmsTransactionManager"/>
<property name="concurrency" value="1-10"/>
<property name="cacheLevelName" value="CACHE_CONSUMER" />

**<property name="sessionTransacted" value="true"/>**
<property name="maxMessagesPerTask" value="1000"/>
<property name="receiveTimeout" value="1000"/>
<property name="destinationName" value="dest.queue"/>
<property name="messageListener" ref="consumptionOrderListener"/>
</bean>

在设置完事务后，再进行测试，两种方式的消费性能就差不多了。题外话，使用事务竟然对性能有这么大的影响，回头再研究下。
鉴于使用bean直接注入DMLC的方式，可以指定maxMessagesPerTask，从而较灵活地伸缩消费线程的数量，最后还是选择了使用bean的方式。
至此，困扰我很长时间的MQ阻塞之谜才终于解开。啊，开心。

参考资料

https://docs.spring.io/spring/docs/3.2.16.RELEASE/spring-framework-reference/htmlsingle/#jms
Why DefaultMessageListenerContainer should not use CachingConnectionFactory? https://stackoverflow.com/questions/21984319/why-defaultmessagelistenercontainer-should-not-use-cachingconnectionfactory
深入理解DefaultMessageListenerContainer http://bijian1013.iteye.com/blog/2309671

探究SpringJMS+ActiveMQ消息阻塞之谜

0x00 背景介绍

0x01 相关概念简介

0x02 探究MQ阻塞之谜

0x02.1 ActiveMQ消费逻辑

0x02.2 基于源码，大胆猜测

0x02.3 通过Eclipse MAT查看对象信息

0x02.4 锁定问题-还是日志大法好

0x03 问题解决

0x03.1 PooledConnectionFactory为啥不缓存MessageConsumer？

0x03.2 简单的性能比对

0x03.03 使用Wireshark和Python Pandas探究性能谜题

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

MQ