最近打算用C#實現一個基於文件的EventStore。

什麼是EventStore

關於什麼是EventStore，如果還不清楚的朋友可以去瞭解下CQRS/Event Sourcing這種架構，我博客中也有大量介紹。EventStore是在Event Sourcing（下麵簡稱ES）模式中，用於存儲事件用的。從DDD的角度來說，每個聚合根在自己的狀態發生變化時都會產生一個或多個領域事件，我們需要把這些事件持久化起來。然後當我們需要恢復聚合根的最新狀態到記憶體時，可以通過ES這種技術，從EventStore獲取該聚合根的所有事件，然後重演這些事件，就能將該聚合根恢復到最新狀態了。這種技術和MySQL的Redo日誌以及Redis的AOF日誌的原理是類似的。但是區別是，redo/AOF日誌是Command Sourcing，而我們這裡說的是Event Sourcing。關於這兩個概念的區別，我不多展開了，有興趣的朋友可以去瞭解下。

為什麼要自己寫一個EventStore

目前ENode使用的EventStore，是基於關係型資料庫SqlServer的。雖然功能上完全滿足要求，但是性能上和數據容量上，離我的預期還有一些距離。比如：

1. 關於性能，雖然可以通過SqlBulkCopy方法，實現較大的寫入吞吐，但是我對EventStore的要求是，需要支持兩個唯一索引：1）聚合根ID+事件版本號唯一；2）聚合根ID+命令ID唯一；當添加這兩個唯一索引後，會很大影響SqlBulkCopy寫入數據的性能；而且SqlBulkCopy只有SqlServer才有，其他資料庫如MySQL沒有，這樣也無形之中限制了ENode的使用場景；
2. 關於使用場景，DB是基於SQL的，他不是簡單的幫我們保存數據，每次寫入數據都要解析SQL，執行SQL，寫入RedoLOG，等；另外，DB還要支持修改數據、通過SQL查詢數據等場景。所以，這就要求DB內部在設計存儲結構時，要兼顧各種場景。而我們現在要實現的EventStore，針對的場景比較簡單：1）追求高吞吐的寫入，沒有修改和刪除；2）查詢非常少，不需要支持複雜的關係型查詢，只需要能支持查詢某個聚合根的所有事件即可；所以，針對這種特定的使用場景，如果有針對性的實現一個EventStore，我相信性能上可以有更大的提升空間；
3. 關於數據量，一個EventStore可能需要存儲大量的事件，百億或千億級別。如果採用DB，那我們只能進行分庫分表，因為單表能存儲的記錄數是有限的，比如1000W，超過這個數量，對寫入性能也會有一定的影響。假設我們現在要存儲100億事件記錄，單表存儲1000W，那就需要1000個表，如果單個物理庫中分100個表，那就需要10個物理庫；如果將來數據量再增加，則需要進一步擴容，那就需要牽涉到資料庫的數據遷移（全量同步、增量同步）這種麻煩的事情。而如果是基於文件版本的EventStore，由於沒有表的概念了，所以單機只要硬碟夠大，就能存儲非常多的數據。並且，最重要的，性能不會因為數據量的增加而下降。當然，EventStore也同樣需要支持擴容，但是由於EventStore中的數據只會Append寫入，不會修改，也不會刪除，所以擴容方案相對於DB來說，要容易做很多。
4. 那為何不使用NoSQL？NoSQL一般都是為大數據、可伸縮、高性能而設計的。因為通常NoSQL不支持上面第一點中所說的二級索引，當然一些文檔型資料庫如MongoDB是支持的，但是對我來說是一個黑盒，我無法駕馭，也沒有使用經驗，所以沒有考慮。
5. 從長遠來看，如果能夠自己根據自己的場景實現一個有針對性的EventStore，那未來如果出現性能瓶頸的問題，自己就有足夠的能力去解決。另外，對自己的技術能力的提高也是一個很大的鍛煉機會。而且這個做好了，說不定又是自己的一個很好的作品，呵呵。所以，為何不嘗試一下呢？

EventStore的設計目標

• 要求高性能順序寫入事件；
• 要求嚴格判斷聚合根的事件是否按版本號順序遞增寫入；
• 支持命令ID的唯一性判斷；
• 支持大量事件的存儲；
• 支持按照聚合根ID查詢該聚合根的所有事件；
• 支持動態擴容；
• 高可用（HA），需要支持集群和主備，二期再做；

EventStore核心問題設計方案

寫入每一個事件時需要保證兩個業務規則

首先我們先看一下每次寫入一個事件時，客戶端會傳給我們什麼信息：

• 聚合根ID
• 事件版本號
• 命令ID
• 事件內容
• 事件發生時間

針對上面的設計目標，寫入一個事件到EventStore時，我們需要保證兩個業務規則：1）當前事件的版本號必須是聚合根的當前版本號（前一個寫入的事件的版本號）+1；2）命令ID唯一；

為什麼要保證這兩個業務規則呢？

第一個是為了能支持檢測聚合根在併發更新時產生的併發衝突，當同一個聚合根在兩個線程中同時被更新（雖然ENode基本保證了不會出現這種情況，但設計上沒有做到絕對的避免），則這兩個更新所產生的事件的版本號是一樣的，而這種情況是不允許的，同一個聚合根的修改必須線性修改。所以EventStore需要能檢測出來這種情況，並告訴客戶端；

第二個是為了能夠自動檢測出同一個CQRS的命令是否重覆執行了，也就是為了命令的冪等處理。因為現代的大部分分散式消息隊列如kafka, rocketmq, rabbitmq, equeue都無法保證消息不會重覆投遞，所以，任何一個命令都有可能被重覆執行，當一個命令被一先一後被執行兩次，然後產生兩個事件，雖然此時這兩個事件的版本號都是沒問題的，但是因為重覆執行了命令，所以很可能會導致最後的結果不正確。所以需要在底層的數據存儲層面檢測出這種情況，並返回給客戶端。通常如果使用DB，針對上面這兩個業務規則，我們可以建立兩個唯一索引即可。

當然，也許你會說，我們可以把這兩個業務規則交給上層應用保證啊，不一定必須在EventStore中做掉。確實，上層應用也可以做，但上層應用因為是無狀態的，而上面這兩個業務規則的檢查都需要依賴於狀態；另外一個原因，上層應用都是集群部署的，所以，如果要由上層自己保證，那必須要用到類似於分散式鎖的東西，整個架構的性能立馬下降一個檔次。

那如何保證這兩個業務規則呢？

第一個業務規則的思考：

很容易想到，我們必須保存當前聚合根的最新版本，這樣在下一個事件過來時，才能判斷出下一個事件的版本是否是當前版本+1。針對這個問題，基於C#語言，最容易想到的就是，我們可以在本地托管記憶體中維護一個ConcurrentDictionary<string, uint>這樣的字典。其中key為聚合根ID，value為聚合根的當前版本號。這樣當一個事件過來時，就能實現上述的判斷了。但是，假設單台EventStore上有1億個聚合根，那就意味著這個字典中就有1億個key，這樣這個字典就會占用不少的記憶體，初步估算了一下，至少有4GB吧。在這麼大的記憶體占用下，GC很可能會有問題。

那怎麼辦呢？另一個方案是使用非托管記憶體來存儲這個字典。但是非托管記憶體中如何實現一個這樣的字典我沒太多經驗，不是C++出生，呵呵。會的同學可以幫我想想怎麼實現，這個對於C++開發來說，應該是比較簡單的需求吧。

關於解決GC的問題，我覺得還有一個辦法也許可行，但我還沒做充分測試，大伙有經驗的也可以幫我看看。思路是：

設計一個環形數組，數組的大小在EventStore啟動時進行初始化，比如為1KW。然後數組中每個元素為一個對象（假設叫VersionEntry），該對象中有兩個欄位：聚合根ID、聚合根當前版本號；然後，當一個聚合根的事件過來時，我們根據聚合根的ID的hashcode取摸環形數組的大小，就能知道該聚合根在數組中的下標了，然後根據下標把VersionEntry拿到，然後判斷VersionEntry中的聚合根ID是否和當前聚合根的ID相同，如果相同，說明當前聚合根的最新版本號在這個環形數組中找到了；如果不相同，則認為沒找到。然後，如果找到的情況下，就更新最新版本號為下一個版本號；如果沒找到，則需要從磁碟嘗試載入該聚合根的最新版本號，這個問題下麵會講到如何實現。

通過這個設計，我們將一定數量的聚合根的最新版本號緩存在一個巨大的數組中，然後EventStore啟動時，就預先初始化好整個數組中的所有對象，當然，此時這些對象的聚合根ID和版本號都是空的。通過之前學習NFX的源碼，我相信通過這樣的數組，可以極大程度的降低Full GC的耗時代價，因為g2沒有任何記憶體碎片，不需要壓縮移動。另外，關於這個環形數組，還有一個優化點，就是hashcode可以支持二級。就是當一級hashcode對應的VersionEntry已經存在且聚合根ID和當前聚合根ID不相同時，自動將該VersionEntry的位置替換為一個新的VersionEntry數組，數組大小不需要太大，比如為7。然後把新老聚合根的信息放入這個新的VersionEntry中，當然，即便是二級hash，還是有可能出現哈希碰撞衝突，此時就覆蓋老數據即可。另外，還有一點比較重要，環形數組的大小應該是質數。

還有最後一點需要再強調一下：

不是說當前的EventStore機器上存儲了1億個聚合根的事件，我們的字典或者環形數據就必須要保存1億個key。我們應該根據實際伺服器的記憶體大小以及GC的影響，來綜合判斷應該緩存多少聚合根。當然，我作為框架設計者，在設計這個緩存方案時，會儘力確保在緩存非常多的key的時候，也沒有什麼大的副作用，比如GC。也就是說，儘量在軟體層面做到無瓶頸，儘量能支持到只要物理記憶體大小足夠，就能支持配置多少大的緩存要求。

第二個業務規則的思考：

第二個業務規則，是一個典型的kv的需求場景，而且我們只需要使用嵌入式的kv即可。兩個選擇：1）自己實現一個；2）使用開源的成熟的高性能嵌入式的kv，如leveldb,stsdb；經過考慮後，還是選擇使用方案二。主要是我覺得既然有成熟的東西可以使用，就應該使用，而不是自己造輪子。目前暫定使用leveldb，當然具體使用哪個還需要進一步調研。

key的設計：命令ID作為key即可。處理邏輯：一個事件過來時，判斷命令ID是否重覆，如果重覆，就直接返回告訴客戶端命令重覆了；否則繼續往下處理。

如何高性能寫入事件以及事件索引？

如何存儲數據？

我們需要存儲的數據有三種，如下：

1. 事件本身數據，寫入到數據文件。單個數據文件的大小固定，比如每個文件1GB。寫入方式為二進位數據順序寫文件，一個文件寫滿後，新建下一個文件，繼續順序寫到新文件；寫數據文件時不需要做任何業務規則檢查，只管寫二進位數據即可。這個設計和EQueue中存放消息的文件一樣，本文就不多做介紹了。有興趣的朋友可以看看這篇文章：http://www.cnblogs.com/netfocus/p/4927495.html
2. 事件索引數據，記錄每個聚合根的每個版本對應的事件在數據文件中的物理位置；有了這個索引數據，我們就能實現根據某個聚合根ID獲取該聚合根的所有版本的事件的需求了；先查索引數據獲取該聚合根的所有版本的事件在數據文件中的物理位置，再根據這些位置最終拿到事件信息。那事件索引數據如何存放呢？也是通過leveldb即可，key為aggId_version，即聚合根ID+聚合根版本號。value為該版本的事件在數據文件中的物理位置；
3. 命令ID數據，我們需要記錄所有的事件的命令ID，這樣才能當一個事件過來時，判斷該事件對應的命令是否已經處理過。這個同樣使用leveldb即可，命令ID作為key。

當一個事件過來時的處理邏輯：

1. 先判斷命令是否被處理過：到leveldb查找key是否存在，判斷命令是否已被處理過；如果已被處理過，則直接返回該事件已被處理過的結果給客戶端；
2. 判斷事件版本號是否合法：如果命令未被處理過，則判斷當前事件的版本號是否是當前聚合根的當前版本號的下一個版本號；上面介紹第一個業務規則時，我們瞭解到，聚合根的當前版本號很可能在緩存（環形數組）里，如果在，則直接可以拿出來判斷；如果不在，則需要從leveldb載入當前版本號。那載入哪些版本號呢？舉個例子來說明吧：假設當前事件的版本號為10，則從事件索引leveldb中嘗試獲取版本號為9的事件以及10的事件。如果存在10的事件，則說明遇到併發衝突了，直接返回客戶端結果告訴客戶端併發衝突；如果10不存在，則繼續判斷9是否存在，如果存在，則符合預期，也就是第一條業務規則滿足條件。如果9也不存在，則認為當前事件的版本號非法，也返回客戶端相應結果即可；因為如果當前聚合根的當前版本號為8，那是不可能過來一個版本號為10的事件的，過來的一定是9，因為聚合根的版本號總是按一依次遞增的。
3. 如果命令和事件版本都合法，就開始寫入數據：1）寫入事件到事件數據文件，2）寫入事件索引到leveldb，3）寫入命令ID數據到leveldb；
4. 三種數據都寫完成功後，更新緩存中當前聚合根的當前版本號為當前事件的版本號；

性能分析：

當一個事件過來時，我們一般是需要訪問三次IO，1）順序寫事件到事件數據文件；2）寫入事件索引到leveldb；3）寫入命令ID到leveldb；大家覺得這3個寫入操作，最終可以提供多少的寫入TPS？我的目標是單機能支持50000TPS。大家覺得這個設計能否做到呢？還是等待最終開發完成後進行測試吧。

如何支持查詢？

除了事件數據的寫入，我們還要支持如何根據聚合根ID獲取該聚合根的所有的事件這個需求。有了前面的介紹，這個問題就很好解決了。當一個查詢請求過來時，我們只需要根據聚合根ID嘗試獲取該聚合根的所有事件即可。首先構造第一個key，aggId_1到事件索引的leveldb中去查找該key是否存在，意思是去嘗試獲取該聚合根的第一個版本的事件的在數據文件中的物理地址。如果存在，就繼續獲取第二個版本的事件，以此類推，直到某個版本不存在事件索引，那就表示該聚合根的所有的事件都獲取到了，就可以返回了。這個查詢基於一個前提就是任意一個聚合根的所有的事件版本都是從1開始，並且總是按1依次遞增的。

如何解決多線程併發寫的時候的CPU占用高的問題？

到這裡，我們分析瞭如何存儲數據，如何寫入數據，還有如何查詢聚合根的所有事件，應該說基本功能已經實現了。另外，如果是單線程訪問EventStore，我相信性能不會很低了。但是如果是N多客戶端同時併發寫事件呢？這個時候就會導致EventStore伺服器會有很多線程要求同時寫入事件到數據文件，但是大家知道寫文件必須是單線程的，如果是多線程，那也要用鎖的機制，保證同一個時刻只能有一個線程在寫文件。最簡單的辦法就是寫文件時用一個lock搞定。但是經過測試發現簡單的使用lock，在多線程的情況下，會導致CPU很高。因為每個線程在處理當前事件時，可能需要涉及到多次IO，所以鎖的占用時間比較長，導致很多線程都在阻塞等待。

為瞭解決這個問題，做了一些調研，最後決定使用雙緩衝隊列的方式來解決。大致思路是：

設計兩個隊列，將要寫入的事件先放入隊列1，然後當前要真正處理的事件放在隊列2。這樣就做到了把接收數據和處理數據這兩個過程在物理上分離，先快速接收數據並放在隊列1，然後處理時把隊列1里的數據放入隊列2，然後隊列2里的數據單線程線性處理。這裡的一個關鍵問題是，如何把隊列1里的數據傳給隊列2呢？是一個個拷貝嗎？不是。這種做法太低效。更好的辦法是用交換兩個隊列的引用的方式。具體思路這裡我不展開了，大家可以網上找一下相關概念。這個設計我覺得最大的好處是，可以有效的降低多線程寫入數據時對鎖的占用時間，本來一次鎖占用後要直接處理事件的，而現在只需要把事件放入隊列即可。雙緩衝隊列可以在很多場景下被使用，我認為，只要是多個消息生產者併發產生消息，然後單個消費者單線程消費消息的場景，都可以使用。而且這個設計還有一個好處，就是我們可以單線程批量處理隊列2里的數據。

如何擴容？

我們再來看一下最後一個我認為比較重要的問題，就是如何擴容。

雖然我們單台EventStore機器只要硬碟夠大，就可以存儲相當多的事件。但是硬碟再大也有上限，所以擴容的需求總是有的。所以如何擴容呢？上面我提到，持久化的數據有三種，經過分析後發現，其實要擴容的數據只有第一種，即事件數據本身。因為事件里包含了所有的信息，聚合根ID，命令ID，事件版本號等。可以說，有了事件數據，我們就能得到另外兩種數據了。歸根結底，另外兩種數據只是事件的兩種二級索引。事件索引數據是根據聚合根ID對事件建立索引；命令ID數據是根據命令ID對事件建立索引。所以，基於這個前提，擴容就很簡單了，我們只需要將事件數據進行擴容即可。

那如何擴容呢？假設現在有4台EventStore機器，要擴容到8台。

有兩個辦法：

1. 土豪的做法：準備8台全新的機器，然後把原來4台機器的全部數據分散到新準備的8台機器上，然後再把老機器上的數據全部刪除；
2. 屌絲的做法：準備4台全新的機器，然後把原來4台機器的一半數據分散到新準備的4台機器上，然後再把老機器上的那一半數據刪除；

對比之下，可以很容易發現土豪的做法比較簡單，因為只需要考慮如何遷移數據到新機器即可，不需要考慮遷移後把已經遷移過去的數據還要刪除。而EventStore的數據是不允許刪除的，只允許追加寫。所以，我放棄了第二種做法。所以，接下來只需要考慮如何實現第一種做法。大體的思路是：

1. 採用拉的方式，新的8台目標機器都在向老的4台源機器拖事件數據；目標機器記錄當前拖到哪裡了，以便如果遇到意外中斷停止後，下次重啟能繼續從該位置繼續拖；
2. 每台源機器都掃描所有的事件數據文件，一個個事件進行掃描，掃描的起始位置由當前要拖數據的目標機器給出；
3. 每台目標機器該拖哪些事件數據？一種可行的方法是：預先在源機器上配置好這次擴容的目標機器的所有唯一標識，如IP；然後當某一臺目標機器過來拖數據時，告知自己的機器的IP。然後源機器根據IP就能知道該目標機器在所有目標機器中排第幾，然後源機器就能知道應該把哪些事件數據同步給該目標機器了。舉個例子：假設當前目標機器的IP在所有IP中排名第3，則針對每個事件，獲取事件的聚合根ID，然後將聚合根ID hash 取摸8，如果餘數為3，則認為該事件需要同步給該目標機器，否則就跳過該事件；通過這樣的思路，我們可以保證同一個聚合根的所有事件都最終同步到了同一臺新的目標機器。只要我們的聚合根ID夠均勻，那最終一定是均勻的把所有聚合根的事件均勻的同步到目標機器上。
4. 當目標機器上同步過來一條事件數據時，同時更新leveldb中的事件索引數據和命令ID數據；

擴容過程的數據同步遷移的思路差不多了。但是擴容過程不僅僅只有數據遷移，還有客戶端路由切換等。那如客戶端何動態切換路由信息呢？或者說如何做到不停機動態擴容呢？呵呵。這個其實是一個外圍的技術。只要數據遷移的速度跟得上數據寫入的速度，然後再配合動態推送新的路由配置信息到所有的客戶端。最終就能實現動態庫容了。這個問題我這裡先不深入了，搞過資料庫動態擴容的朋友應該都瞭解原理。無非就是一個全量數據遷移、增量數據遷移、數據校驗、短暫停止寫服務，切換路由配置信息這幾個關鍵的步驟。我上面介紹的是最核心的數據遷移的思路。

結束語

本文介紹了我之前一直想做的一個基於文件版本的EventStore的關鍵設計思路，希望通過這篇文章把自己的思路系統整理出來。一方面通過寫文章可以進一步確信自己的思路是否OK，因為如果你文章寫不出來，其實思路一定是哪裡有問題，寫文章的過程就是大腦整理思緒的過程。所以，寫文章也是檢查自己設計的一種好方法。另一方面，也可以通過自己的原創分享，希望和大家交流，希望大家能給我一些意見或建議。這樣也許可以在我動手寫代碼前能及時糾正一些設計上的錯誤。最後再補充一點，語言不重要，重要的是設計思路。誰說C#語言做不出好東西呢？呵呵。