CQRS的事件總線

Question

我用CQRS編寫一個項目。我想從域事件異步構建我的讀取模型。當我閱讀Greg Young和其他關於CQRS的話題時，我看到的只是使用服務總線。但對我來說，服務總線是消息傳輸的技術層，在這種情況下，處理的順序並不明顯。

域事件是具有特殊性的消息：處理順序是顯性的。因此，我使用服務總線（NServiceBus，Mass transit ...）和額外的（複雜）層來確保事件處理順序嗎？或者具有這種特殊性的特定「事件公共汽車」已經存在？

編輯

我解釋更準確地說我的情況：

我做了CQRS第一應用。在這個應用程序中，我使用關係數據庫作爲持久層而不是事件源。該數據庫用於寫入側和讀取側。這是一個選擇。此應用程序在沒有實現的抽象總線上發佈域事件。

我需要使用第一個應用程序的讀取模型構建第二個應用程序。

在這種情況下，如果我使用事件存儲，我使用和事件存儲爲我的第二個應用程序構建投影是一件壞事。如果持久層發生更改，則其他應用程序（偵聽器）會受到影響。

編輯2：解決方案？

解決方案可能是使用兩個事件存儲/服務總線來構建事件總線。因此，持久性的無知被保留下來並且事件總線成爲建立投影和聽事件的工具。

編輯3：GetEventStore

我看GetEventStore，它提供了一個訂閱新機制。 解決方案可能使用GetEventStore作爲事件總線！

因此，如果我使用GetEventStore作爲事件總線，這意味着我使用事件源？或者事件源只是一個僅用於持久層的概念？或者在我的第一個應用程序中，我使用了事件源，因爲我使用聚合的域事件來構建關係數據庫中的狀態？

所以新的問題是什麼是事件採購？將事件直接存儲在事件存儲中或使用域事件來構建狀態？

Answer 1

這裏有幾件事要考慮。

如果您使用事件採購，則可能需要使用GES（geteventstore.com）。在這種情況下，用於構建讀取模型的事件不通過任何消息總線，而是使用訂閱從事件存儲中檢索。在這種情況下，你總是會得到適當的訂單。

如果你在「普通」數據庫中只有兩個單獨的模型，這是一個文檔還是關係數據庫，你可能應該使用一些消息總線。這裏還有一些東西：

• 請記住，更新寫入端並將事件發佈到消息總線應該在事務中完成。否則，你可能冒着讀取模型不一致的風險

• NServiceBus或MassTransit沒有作爲服務總線抽象來處理消息排序。事實上，你應該選擇合適的運輸工具，並完成這項工作。例如，AMQP爲消息排序和RabbitMQ delivers設置了一些要求。當然，使用NSB或MT會讓你的生活變得更容易，但它是完全不同的東西

• 通過將聚合版本保持爲兩個字段，您始終可以保持您的寫入和讀取方在同一版本上兩側。在這種情況下，如果你得到一個空白的一些事件，並決定該怎麼做，你至少可以知道 - 等待交付丟失的事件或只是停止處理和崩潰，讓人工干預

實踐表明，所有現代經紀人都支持AMQP並遵循標準。儘管你可能有重複，但這樣做危險性較小。

你可以選擇使用ZMQ，這會給你幾乎爲零的延遲，所以傳遞消息的順序的變化將是微不足道的，除非你的卷是非常高的。

Answer 2

所以新的問題是什麼是事件採購？將事件直接存儲在事件存儲中或使用域事件來構建狀態？

「事件來源」是所有的更改您的域模型域事件被捕獲的屬性。

域事件是具有特殊性的消息：處理順序是顯性的。

是的，這很難理解;如果你正在處理來自的任何事件，你需要閱讀歷史（有序的事件序列），而不僅僅是事件本身。

我認爲使用事件存儲（在選擇事件源的情況下）使讀取模型的預測意味着與持久層的依賴性很差。

完全沒有;從讀取模型的角度來看，只有一個存儲庫返回歷史記錄。實際的實現細節純粹是一個持久性問題。

這樣，我用一個服務總線（NServiceBus，公共交通...）與其它附加（復）層，以保證事件處理順序

幾乎 - 你使用任何事件發佈你像簡單的圖層來確保事件處理順序。

我想從域事件異步構建我的讀取模型。

因此，您的訂閱者在查看每個事件時會查看它所關注的事件，並跟蹤哪些歷史記錄已過時。當重建模型的時候，您重新加載已更改的歷史記錄（或者如果您一直保持跟蹤，則獲取這些歷史記錄的更新）。

如果你的事件有版本信息編碼到它們中，你可以做得更好一些，因爲你可以跟蹤你在流中的位置;並且知道你已經處理過他們的任何「舊」事件。

也就是說，想象你的用戶在更新的地圖中，其跟蹤每個流/歷史/骨料最新的事件是看

{ user.7 : 14 
, order.14 : 27 
, order.12 : 4 
} 

當它的時間來重建讀取模型的內存拷貝，您複製該地圖，並將其交給將構建模型的過程。該進程具有自己的映射，描述了用於重建以前版本的流的版本。

{ user.7 : 14 
, order.14 : 22 
, order.12 : 5 
} 

這裏，過程可以看到，前面的投影是最新的user.7的背後，是對order.14，並提前對order.12（也許是用戶錯過的事件，或者它不是招還沒到，或者什麼的）。

既然我們知道狀態是狀態，我們重建讀取模型;你可以從頭開始重建所有東西，或者你可以注意到只有order.14是過時的，並從那裏重建。獲取您需要的歷史記錄，創建讀取模型，保存用於創建該模型的流版本，然後完成。

如果第一個應用程序更改其持久層，則其他應用程序會受到影響，不是嗎？

不一定 - 如果你想要我的數據，那麼你跟我說話 - 而不是我的持久層。你應該回顧一下Udi Dahan對於[特定業務能力] [2]的technical authority服務的說法。

Answer 3

事件採集和消息傳遞之間存在根本區別。

在消息系統中，消息在很大程度上是暫時的。一些排隊系統在將隊列從隊列中拉出後仍然保留消息，但邏輯上它們已經被處理並且可以被遺忘。

對於事件源，您明確地設計了從域模型發出的事件以表示聚合的狀態。這些結構中的數據通常會與使用服務總線發送的系統間消息有所不同。這僅僅是因爲您可能擁有比使用事件採購恢復聚合狀態所需的系統消息更多的信息。

您ES店中的事件始終按正確的順序排列。這應該使用聚合/流的版本來強加。一個用戶不應該「覆蓋」另一個用戶的事件，甚至不可能。

爲了重新構建您的聚合體，請使用ES商店。你不能也不應該依賴服務總線式的消息。

要構建您使用事件存儲的投影。由於這些事件的順序是正確的，所以你不需要特別的閱讀。事件存儲應該有一定的序列跨越所有aggregtes：

| aggregate id | data  | version | sequence number | 
| ------------ | -------- | ------- | --------------- | 
| 521   | {binary} | 1  | 1    | 
| 521   | {binary} | 2  | 2    | 
| 521   | {binary} | 3  | 3    | 
| 516   | {binary} | 1  | 4    | 
| 516   | {binary} | 2  | 5    | 
| 7221   | {binary} | 1  | 6    | 

即使事件流的版本是每個聚集的序列號是在所有獨特。通過這種方式，你可以爲投影讀取事件從序號1開始

然後，您可以有很多的預測，因爲你需要每個一些位置指示器：

| projection | position | 
| ---------- | -------- | 
| customer | 2  | 
| order  | 6  | 

你的事件投影處理器將繼續閱讀事件並將它們交給處理程序處理。處理程序將更新讀取模型。位置光標將被移動。

當您發現您在投影上發生了某些處理錯誤時，這非常方便。您可以刪除您的讀取模型並將位置重置爲0以重新處理整個事件歷史記錄。

隨時添加新投影也很容易。

我希望有幫助。如果您想尋找一些靈感，我有一個名爲Shuttle.Recall的事件採購機制:)