ActiveMQ持久化消息

ActiveMQ可以将消息持久化到后端存储,像KahaDB,JDBC等。

遍历持久化消息

ActiveMQ提供了存储级消息迭代器来遍历持久化消息:

// 获取连接、会话

PersistenceAdapter adapter = 
    (PersistenceAdapter) broker.getAdaptor(destination);

MessageCursor cursor = adapter.browse();

while (cursor.hasNext()) {
  Message message = cursor.next();   

   // 处理消息   
}

MessageCursor实现了数据迭代器,next()方法返回下一个消息。

ActiveMQ通过实现迭代器接口来遍历存储的消息:

MessageCursor 实现了java.util.Iterator接口
支持hasNext()和next()方法
可以遍历 underlying store 中的消息
一次只加载一个消息对象到内存,节省空间
支持任意方向的迭代。

内部实现

在具体的 PersistenceAdapter 中,实现 MessageCursor 接口:

public class KahaPersistenceAdapter extends PersistenceAdapter {

  public MessageCursor browse() {     
     return new KahaMessageCursor(...);
  }     

}  

public KahaMessageCursor implements MessageCursor {
   //  存储物理位置
   ... 

   public boolean hasNext() { ..}    
   public Message next() { ..}
}

这里会根据不同持久化实现,返回不同的MessageCursor。

利用迭代器

ActiveMQ通过提供存储级消息迭代器,可以:

遍历所有存储的消息
按序处理消息
支持任意方向遍历
加载消息对象的同时节省空间

符合迭代器模式,解耦了数据的逻辑和物理存储。