EventStore文件存储设计

背景

enode是一个cqrs+event sourcing架构的开发框架，event sourcing需要持久化事件，事件可以持久化在db，但是db由于面向的是crud场景，是针对数据会不断修改或删除的场景，所以内部实现会比较复杂，性能也相对比较低。而event store实际上对数据只有新增和查询的需求，所以我想为event sourcing的场景针对性的实现一个event store。看了一下业界的一些实现，感觉都没有达到我的期望，所以想自己动手实现一个。下面是我构思的一个event store的单机版应该要具备的能力以及对应的设计方案，分享出来和大家讨论。

一、需求概述

• 存储聚合根的事件数据
• 支持事件的版本并发控制，新事件的版本号必须是当前版本号+1
• 支持命令重复判断，即不可以处理重复命令产生的事件
• 支持按聚合根id查询该聚合根的所有事件
• 支持按聚合根id+事件版本号查询指定的事件
• 支持按命令id查询该命令对应的事件数据
• 高性能，写入要尽量快，查询要尽量快

二、事件数据格式

{
  "aggregaterootid": "",     //聚合根id
  "aggregateroottype": "",   //聚合根类型
  "eventversion": "",        //事件版本号
  "eventtime": "",           //事件发生时间
  "eventdata": "",           //事件数据，json格式
  "commandid": "",           //产生该事件的命令id
  "commandtime": ""          //产生该事件的命令产生时间
}

三、存储设计

1、核心内存存储设计

• 遵循内存只存储索引数据的原则，尽量充分利用内存；
• aggregatelatestversiondict，存储每个聚合根的最大事件版本号
  • key：aggregaterootid，聚合根id
  • value：
    • eventversion，当前聚合根的最新事件的版本号，也即当前聚合根的版本号
    • eventtime，事件产生时间
    • eventposition，事件在事件数据文件中的位置
• commandiddict，存储命令索引
  • key：commandid，命令id
  • value：
    • commandtime，命令产生时间
    • eventposition，命令对应的事件在事件数据文件中的位置

2、物理存储的数据

• 事件数据：eventdata，单条数据的结构：

{
  "aggregaterootid": "",     //聚合根id
  "aggregateroottype": "",   //聚合根类型
  "eventversion": "",        //事件版本号
  "eventtime": "",           //事件发生时间
  "eventdata": "",           //事件数据，json格式
  "commandid": "",           //产生该事件的命令id
  "commandtime": "",         //产生该事件的命令产生的事件
  "previouseventposition": ""//前一个事件在事件文件中的位置
}

• 事件索引：eventindex，单条数据的结构：

{
  "aggregaterootid": "",     //聚合根id
  "eventversion": "",        //事件版本号
  "eventtime": "",           //事件产生时间
  "eventposition": "",       //事件在事件数据文件中的位置
}

• 命令索引：commandindex，存储内容：存储所有命令的id及其对应的事件所在文件的位置

{
  "commandid": "",        //聚合根id
  "commandtime": "",      //命令产生时间
  "eventposition": "",    //事件在事件数据文件中的位置
}

3、事件数据存储

• 同步顺序写eventdatachunk文件，一个文件大小为1gb，写满一个文件后写入下一个文件；
• 写入每个事件时，同时写入当前事件的前一个事件所在的文件位置，以便将来可以一次性将某个聚合根的所有事件从文件查找出来；

4、事件索引存储

• 异步顺序写eventindexchunk文件，一个文件大小为1gb，写满一个文件后写入下一个文件；
• 对于已经写满的不会再变化的文件的内容，使用后台线程进行b+树索引整理，索引的排序依据是聚合根id+事件版本号；b+树设计为3层，根节点包含1000个子节点，每个子节点再包含1000个子节点，这样叶子节点共有100w个。每个叶子节点我们保存20个版本索引，则单个文件共可保存最多2000w个版本索引，10个文件为2亿个版本索引；单机存储2亿个事件索引，应该可以满足大部分应用场景了；3层，则查找任意一个节点，只需要3次io访问；
• 由于是后台线程对已经写完的文件进行b+树索引整理，b+树是在内存建立，建立完成后，将最新的内容写入新文件，原子替换老的eventindexchunk文件；所以，这块的逻辑处理应该不会对服务的主逻辑产生较大的影响；
• 采用bloomfilter优化查询性能，使用bloomfilter来快速判断某个eventindexchunk文件中是否包含某个聚合根id，如果不在，则不用从b+树去检索该聚合根的版本号了；如果在，则取检索；通过这个设计，当我们要获取某个聚合根的最大版本号时，不需要对每个eventindexchunk文件进行b+树查询，而是先通过bloomfilter快速判断当前的eventindexchunk文件是否包含该聚合根的信息，大大提升检索效率；bloomfilter的二进制bit数据占用内存小，可以在每个eventindexchunk文件被扫描时，和文件头的信息一起加载到内存；

5、命令索引存储

• 异步顺序写commandindexchunk文件，一个文件大小为1gb，写满一个文件后写入下一个文件；
• 同事件索引存储，进行b+树索引建立，索引的排序依据是命令id；
• 同事件索引存储，采用bloomfilter优化查询性能；

四、框架逻辑设计

1、查询某个聚合根的最大版本号

• eventstore启动时，会加载所有的eventindexchunk文件的元数据到内存，比如文件号、文件头、bloomfilter等信息，但不真实加载文件内容，文件数不会太多，最多也就几十个；
• 根据聚合根id+bloomfilter算法，快速确定应该到哪个eventindexchunk文件中去查找该聚合根的最新版本号，eventindexchunk文件从新到旧遍历，因为某个聚合根id的最大版本号一定是在最新的eventindexchunk文件中的；
• 在找到的eventindexchunk中使用b+树查找算法，找到对应的叶子节点；
• 在找到的叶子节点，使用二分查找算法（由于单个节点的聚合根id不多，顺序查找即可），找到指定聚合根的最新版本号；

2、查询某个聚合根的所有事件

• 先通过上面的算法找出该聚合根的最大版本号的事件在事件数据文件中的位置；
• 然后从该位置获取事件完整数据；
• 再根据事件数据中记录的上一个事件在事件数据文件中的位置，查找上一个事件的数据；
• 以此类推，直到找到该聚合根的第一个事件的数据；

3、查询某个命令对应的事件数据

• 先尝试从内存查询该命令的索引信息，如果存在，则直接获取该命令对应的事件在事件数据文件中的位置，即eventposition；如果不存在，则尝试从命令的索引文件中查找，结合bloomfilter和b+树查找算法进行查找；
• 如果找到了eventposition，则根据eventposition到事件数据文件中查找对应的事件数据即可；如果未找到，则返回空；

4、追加一个新事件的处理逻辑

• 根据aggregatelatestversiondict判断事件版本号是否合法，必须是聚合根的当前版本号+1，如果当前版本号不存在，则首先尝试从eventindexchunk文件查找当前聚合根的最大版本号，如果还是查找不到，说明当前聚合根确实不存在任何事件，则当前事件版本号必须为1；
• 根据commandiddict判断命令id是否重复，如果commandiddict中不存在该命令，尝试从commandindexchunk文件中查找，也是b+树的方式；这里需要设计一个配置项，让开发者配置是否需要继续从commandindexchunk文件查找命令id。有时我们只希望从内存查找即可，不希望再从磁盘查找了，因为判断命令是否重复我们很多时候只希望检查最近一段时间内的命令，检查全部命令代价过大，意义也不是很大；
• 如果事件的版本号合法、命令id不重复，则append的方式写入事件数据到eventdatachunk；
• 写入完成后，更新aggregatelatestversiondict、commandiddict，、bloomfilter的bit数组，以及将当前的事件放入内存的一个双缓冲队列；队列消费者异步批量将事件索引和命令索引写入对应的索引文件；
• 返回事件写入结果；

5、其他逻辑

• 异步线程定时批量持久化事件索引；
• 异步线程定时批量持久化命令索引；
• 异步线程定时清理不需要放在内存的聚合根最新版本号信息（aggregatelatestversiondict中的key），根据eventtime判断，只保留最近1周有过变化（产生过事件）的聚合根；
• 异步线程定时清理不需要放在内存的命令索引（commandiddict中的key），根据commandtime判断，只保留最近1周的命令id；
• 异步线程定时进行事件索引和命令索引的b+树索引的建立，即对已经写入完成的eventindexchunk和commandindexchunk文件的内部重构；
• eventindexchunk和commandindexchunk文件标记为写入完成前，要把bloomfilter的bit数组内容写入文件中；
• 其他eventstore的启动逻辑，比如启动时加载一定数量的索引数据到内存，以及索引数据相比事件数据是否有漏掉或无效的检查；
• 其他逻辑支持，如支持聚合根的快照存储，从文件查找数据时，如果文件的b+树索引信息还未建立，则需要进行全文扫码；