ActiveMQ之消息指针

Posted 2020-06-13

 

　　消息指针(Message cursor)是activeMQ里一个非常重要的核心类，它是提供某种优化消息存储的方法。消息中间件的实现一般都是当消费者准备好消费消息的时候，它会从持久化存储中一批一批的读取消息，并发送给消费者。消息指针维护着下一批待读取消息的相关位置信息。 

消息游标:

　　当producer发送的持久化消息到达broker之后，broker首先会把它保存在持久存储中。接下来，如果发现当前有活跃的consumer，而且这个consumer消费消息的速度能跟上producer生产消息的速度，那么ActiveMQ会直接把消息传递给broker内部跟这个consumer关联的dispatch queue；如果当前没有活跃的consumer或者consumer消费消息的速度跟不上producer生产消息的速度，那么ActiveMQ会使用Pending Message Cussors保存对消息的引用。在需要的时候，Pending Message Cursors把消息引用传递给broker内部跟这个consumer关联的dispatch queue。

 

　　消息指针在对不同的消息消费者时，它的内部处理机制也不一样： 
　　(1) 当消费者跟得上消息生产者的时候，是快消费者。这种时候Message cursor的内部过程如下图所示： 

　　　　

 

　　(2) 当消费者慢于消息生产者的时候，是慢消费者。那这种时候Message cursor的内部过程如下图所示：

　　　　

 

　　上面两种情况是能自动调整的，当一个消费者从快变成慢或从慢变成快的时候，Message cursor应该做自动的调整。

 　　(在5.1里面这种自动调整有点小bug，它只能从快变成慢，反之则不行。具体bug原因应该是疏忽写错了，代码在类AbstractStoreCursor中的public final synchronized void remove()方法中的if (size==0 && isStarted() && cacheEnabled)这一行，只用把cacheEnabled改为useCache就可以了。(该bug已经被后续版本所修复))。

 

ActiveMQ的消息指针(message cursor)分为3种:

(1) Store-based

(2) File-based

(3) VM

 

下面我具体分析每种消息指针的实现：

(1) Store-based实现原理图:

　　

　　消息接收后，首先完成消息存储的工作，如①所示，然后判断是否有空闲的内存可用，如果有的话，就走路径1，直接把消息存放在内存中的LinkedHashMap，如果没有可用内存，则走路径2，当需要消息的时候，直接从消息存储的介质里每次读取一批消息，然后存入LinkedHashMap。

(2) File实现原理图：

　　

　　消息接收后，首先完成消息存储的工作，如①所示，然后判断是否有空闲的内存可用，如果有的话，就走路径1，直接把消息存放在内存中的LinkedList，如果没有可用内存，则走路径2，把消息写入临时文件中，当需要消息的时候，直接从临时文件中读写一批，然后送入LinkedList。

(3) VM实现原理图：   

　　

　　消息接收后，首先完成消息存储的工作，如①所示，然后直接把消息存放在内存中的LinkedList。

  

3种消息指针的比较: 

　　

下面是“追风”提供的消息指针的测试记录:

(1) 基于存储的消息指针_测试结果:

　　

(2) 基于文件的消息指针_测试结果

　　

 (3) VM消息指针_测试结果: