背景

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+樹索引資訊還未建立，則需要進行全文掃碼；