activeMQ 的kahadb存储引擎分析

很久没更新blog了，前几天看到淘宝开源了Meta，估计notify也要开源了。其实消息中间件的一个非常重要的核心部件就是持久化存储，可能Meta的功能定位使得它在这一块的实现相对notify和activemq就简单些。趁着有点时间，把activeMQ的kahadb存储引擎做了个分析，希望能对jms实现感兴趣的朋友有点帮助。
1． 概述
Kahadb是activemq从版本5.4之后的默认消息存储引擎。消息存储机制是消息中间件最重要的核心部件和性能提升点。一直想对它做一个完整分析，这次趁有时间对kahadb做一个较完整分析。
Kahadb是基于B-tree算法的，具体原理fusesource给了个原理说明(http://fusesource.com/docs/broker/5.4/persistence/KahaDB-Overview.html)，下面我们从代码实现角度进行一个较深入的分析。下面所有的分析都是基于activeMQ 5.4.3版本源码，该版本里的kahadb版本是V3，且是基于queue进行的源码分析，topic的实现虽然有不少差异，但整体可参考queue的。

2． 每个磁盘文件的作用和详细存储格式
首先kahadb在消息保存目录中只有4类文件和一个lock，跟ActiveMQ的其他几种文件存储引擎相比这就非常简洁了。
每种文件的具体作用：
#db-*.log:
存放完整的每条消息(包括事务、目的地、id、优先级、具体内容等)和producerSequenceIdTracker(用来验证每个消息生成者发送的消息是否重复的数据结构)。它随着消息数量的增多，如每32M一个文件，文件名按照数字进行编号，如db-1.log、db-2.log、db-3.log …

#db.data:
通过存放多个Btree数据结构来保存各类重要信息，下面一一进行介绍：
 Metadata类的destinations：用来保存该broker上有哪些Queue或队列
 StoredDestination类的orderIndex属性中的
defaultPriorityIndex、lowPriorityIndex、highPriorityIndex，这3个btree是为消息优先级排序而设计的(应该是版本5.4引入的，唉，有时候一个功能的引入带来的代价可能比较大)。它们的主要作用是为AbstractStoreCursor类的doFillBatch方法服务的，也就是常说的消息指针(message cursors)。当消息指针需要从磁盘文件中装载一批消息的时候会使用这3个btree实例(kahadb版本小于2的不支持lowPriorityIndex、highPriorityIndex)
 StoredDestination类的locationIndex：该btree的主要作用包括：
. 系统重启进行恢复操作的时候，要移除掉不在db-*.log文件里的消息；
. 在系统进行定时checkpointUpdate时使用
 StoredDestination类的messageIdIndex：该btree的主要作用是消息确认acknowledge操作时，通过消息ID在messageIdIndex中删除对应的记录，并依据返回的值删除orderIndex和locationIndex中的记录
上面这些就是kahadb中最主用的btree实例。

#db.redo:
它的作用是“Double Write”，具体代码参看PageFile类的writeBatch方法。它的原理可参考(http://www.mysqlperformanceblog.com/2006/08/04/innodb-double-write/)

#db.free:
当前db.data文件里哪些页面是空闲的，文件具体内容是所有空闲页的ID

下面是具体每个文件的内部数据格式：

 

 

看来大家关注底层的比较少哈，最近也在考虑自己做一个分布式queue的存储，支持动态扩容，数据顺序性，数据多写(解决数据不丢)。

lz对这方面的存储，有研究不？

可以考虑下淘宝的 metamorphosis，已经实现了HA复制。

看了下metamorphosis，我的几个需求基本能满足，数据不丢这个基于master/slave模式不一定能100%保证吧, master disk crash了之后，总会丢一些数据。

有对应的性能测试报告不？

最近看了下bookkeeper + hedwig的实现。不过他的两层服务架构部署已经可用性要打点折扣

master/slave有两种模式，同步复制和异步复制，这跟mysql是机制是一样的。不过我更推荐异步复制，同步复制的性能会差一些，并且机制相对复杂。

另外，一般我们磁盘都做RAID，这样一来其实异步复制就可以接近100％的高可用。

性能测试报告内部有做，不过好像没放出来，性能很大程度上取决于磁盘和参数配置，磁盘是SATA还是SAS，参数主要是每隔多少秒force和每隔多少个消息force。总体上说，性能不是个问题。
一个印象数据，2K的消息，100个并发发送消息，每隔10秒force和每1000条消息force，TPS接近25000，磁盘是做RAID 10的SAS盘，15000转。如果是普通SATA盘，性能下降一半多。异步复制对性能没有影响。另外，在设置每条消息做force的情况下，会利用group commit技术来提高吞吐量，TPS不下降，但是CPU会上升很多。
有兴趣可以自己测试下。

看了下文档，存在几个疑问：

1.

我的业务场景是：顺序推送数据到server，如果涉及到分区，那如果单个消费者在处理时，连接到多个分区后。多个分区的处理数据相比于原先的推送顺序，应该不能保持一致吧。

2. metamorphosis没看到topic的相关设计，不知道是不是我看漏了。MessageConsumer在订阅的时候没有传递订阅者id？ 如果一个订阅者短暂型的crash了，对应这段时间产生的数据是否会继续给他保留？

我的业务场景比较特殊，主要是用于记录数据库的binlog/redolog的解析结果，需要保证顺序性，不然消费时进行数据同步，载入目标库语义上就会错乱。

publish性能有个万把tps就够用了，后端的订阅消费速度只有5000tps以下。

1.发送者可以自定义分区选择器来保证需要顺序的消息发到同一个分区。
2.有的，meta同样有topic，consumer有group的概念，也就是你说的id，这个设计跟kafka是一样的，可以看kafka的设计文档。数据的保留跟订阅者没有关系，服务端会按照配置的策略保存消息，客户端什么时候上来都可以，只要消息还没有被服务器删除都可以读到。具体还是看kafka的设计文档。

恩，大致明白了metamorphosis的设计，和我这边的需求有些少许差异。
1. 如果自定义了分区算法，比如需要顺序的分到一个区上，那对应的数据存储的扩展性是不是就没了，取决于一个分区的disk容量。说白了，我这边的所有数据都是有顺序性要求，但又要支持水平扩展性(通过增加机器扩充存储)，又得保证数据不能丢失。
2. 针对订阅者的一个管理，我原先的想法是为每条消息定义一个全局顺序的id，记录每个subscribe消费到哪个id。定时的删除最后一个subscribe的message id之前的数据。也就是说只要有订阅者订阅过一次，subscribe的offest在，那对应的数据就永远不会被删除。不过我们还有个特殊的需求，就是希望数据可以做replay，消费过一次后，再连接上来后replay一次

我之前看了下bookkeeper + hedwig的实现，网络架构上有点复杂了，两层传输，可靠性要打点折扣。