探索cqrs和事件源

關(guān)于事件源和CQRS的討論似乎通常集中在CQRS上下文中的整體系統(tǒng)架構(gòu)或領(lǐng)域驅(qū)動設(shè)計的各種形式。 但是，盡管也有一些有趣的考慮，但讀取模型經(jīng)常被忽略。 在本文中，我們將介紹通過使用事件流來填充視圖模型的示例實現(xiàn)。

總覽

讀取模型的想法非常簡單。 您獲取事件日志，使用適當(dāng)?shù)墓δ茉谧畛鯙榭盏臄?shù)據(jù)模型上應(yīng)用（重放）所有事件，然后獲得填充的模型。 代碼如下所示：

List<Event> events = getEvents(); Model model = Model.empty(); for (Event event : events) {apply(model, event); }

通過函數(shù)式編程，我們可以使此過程更短：

Model m = reduce(getEvents(),Model.empty(),(m, e) -> apply(m, e));

這就是本質(zhì)。 請注意，這只是抽象的輪廓，實際的實現(xiàn)可能會有所不同，包括緩沖，批處理（或流式傳輸），持久性等。

申請活動

應(yīng)用事件的實際Java代碼可能類似于以下內(nèi)容：

EventProcessingResult processEvents() {if (getState().isRunning()) {int batchSize = getEventsPerIteration();List<Event> events = eventStore.getEventsForAllStreams(getLastEventId(),batchSize);if (events.isEmpty()) {return NO_EVENTS_TO_PROCESS;} else {return processEvents(events);}} else {return NOT_RUNNING;} }EventProcessingResult processEvents(List<Event> events) {try {for (Event event : events) {dispatchEvent(event);}return SUCCESS;} catch (RuntimeException e) {return FAILURE;} }

總而言之，這確實非常簡單明了。 在處理單個事件和整個批處理之前和之后，可以使用鉤子來增強它。 這樣的鉤子可以用來：

• 實施交易，
• 插入監(jiān)控，
• 實施錯誤處理，
• 根據(jù)速度計算批次大小，
• 執(zhí)行任意操作，例如設(shè)置某些內(nèi)容或每批重新計算一次。

最后一個有趣的部分是dispatchEvent方法。 除了遍歷類型層次結(jié)構(gòu)，錯誤處理并使其全部可選之外，它還可以歸結(jié)為：

void dispatchEvent(Event e) {Method handler = projector.getClass().findMethod("on", e.getClass());handler.invoke(projector, e); }

換句話說，對于每種事件類型（例如OrderCreated ），我們在projector對象上尋找一個名為on的公共方法on該方法采用一個匹配類型的單個參數(shù)。

以上所有都是引擎的一部分，是支持許多視圖模型的基礎(chǔ)架構(gòu)。 實施投影所需的所有操作實際上是為投影儀提供有趣的事件類型的處理程序。 所有其他事件將被忽略。

它可能看起來像這樣：

public class OrderProjector {@Injectprivate OrderDao orders;public void on(OrderCreated e) {orders.save(new Order(e.getOrderNumber()));}public void on(OrderApproved e) {Order o = orders.find(e.getOrderNumber());o.setApproved(true);} }

投影線

讓我們討論一下多線程。 共享的可變狀態(tài)立即帶來許多問題， 應(yīng)盡可能避免 。 處理它的方法之一是首先沒有并發(fā)性，例如通過限制對單個線程的寫入。 在大多數(shù)情況下 ，單線程寫入器與ACID事務(wù)相結(jié)合足以應(yīng)付寫入負(fù)載。 （讀取/查詢負(fù)載可能很重，并且使用許多線程–此處所有詳細信息僅與寫入有關(guān)。）

從查詢事件存儲到更新視圖模型數(shù)據(jù)庫，該線程負(fù)責(zé)將事件應(yīng)用于讀取的模型。 通常，它只是從商店加載一批事件并應(yīng)用它們。 只要有更多事件要處理，它就會繼續(xù)，并在捕獲到事件后進入睡眠狀態(tài)。 在一定時間后或事件存儲通知新事件時，它將喚醒。

我們還可以控制該線程的生命周期。 例如，我們提供了一種以編程方式暫停和恢復(fù)每個投影線程的方法，甚至可以在管理GUI中公開。

推還是拉？

使用數(shù)據(jù)庫支持的事件存儲，可以很容易地重復(fù)查詢新事件。 這是拉模型。 不幸的是，這也意味著您可能最終會輪詢過多并產(chǎn)生不必要的負(fù)載，或者輪詢頻率太低，因此可能需要更長的時間才能將更改傳播到視圖模型。

這就是為什么除了輪詢事件存儲之外，最好引入通知，以在保存新事件后立即喚醒讀取的模型。 這有效地成為了具有最小延遲和負(fù)載的推送模型。 我們發(fā)現(xiàn)JGroups是完成這項工作的非常好的工具-它支持多種協(xié)議，并且易于設(shè)置，與成熟的消息隊列相比，所涉及的麻煩要少得多。

通知可能包含也可能不包含實際事件。

在后一種（和更簡單的）設(shè)計中，它們僅傳播已保存新事件的信息及其順序ID（以便所有預(yù)測都可以估算出其背后的數(shù)量）。 喚醒后，執(zhí)行程序可以從查詢事件存儲開始沿其正常路徑繼續(xù)。

為什么？ 因為處理來自單一來源的事件比較容易，但是更重要的是，由于數(shù)據(jù)庫支持的事件存儲可輕松保證排序，并且消息丟失或重復(fù)不存在任何問題。 鑒于我們正在按主鍵順序讀取單個表，而且大多數(shù)情況下數(shù)據(jù)仍在RAM高速緩存中，因此查詢數(shù)據(jù)庫的速度非常快。 瓶頸在投影線程中，正在更新其讀取模型數(shù)據(jù)庫。

但是，將事件數(shù)據(jù)放入通知中沒有任何障礙（可能是出于大小或網(wǎng)絡(luò)流量方面的考慮）。 這可能會減少事件存儲上的負(fù)載并節(jié)省一些數(shù)據(jù)庫往返時間。 投影儀將需要維護一個緩沖區(qū)，并在需要時退回查詢事件存儲。 或者系統(tǒng)可以使用更可靠的消息隊列。

重新開始投影

除了暫停/恢復(fù)之外，上面的屏幕快照還顯示了另一項操作：重新啟動。 從外觀上看是無害的，這是一個非常不錯且功能強大的功能。

由于視圖模型是完全從事件日志派生的，因此可以隨時丟棄它并從頭開始創(chuàng)建（或從某個初始狀態(tài)/足夠老的快照開始）。 在事件日志中，數(shù)據(jù)是安全的，這是事實的最終來源。

當(dāng)有關(guān)視圖的任何內(nèi)容發(fā)生更改時，此功能很有用：添加了字段或表，修復(fù)了錯誤，計算出的內(nèi)容有所不同。 當(dāng)發(fā)生這種情況時，通常從一開始就更容易（或需要），而不是例如實施大量SQL遷移腳本。

甚至可以進行完全自動化，以便在系統(tǒng)啟動并檢測到DB模式與相應(yīng)的Java模型不匹配時，它可以自動重新創(chuàng)建模式并重新處理事件日志。 就像使用Hibernate create-drop策略運行一樣，不同之處在于它不會丟失數(shù)據(jù)。

性能

該解決方案在性能方面可能看起來非常有限。

單線程作家可能引起人們的注意。 實際上，單個線程通常足夠快，可以輕松地跟上負(fù)載。 并發(fā)不僅更難以實現(xiàn)和維護，而且還引入了競爭。 讀取（查詢）可以是多線程的，并且易于擴展。

通過擁有多個讀取模型，例如從管理和“交易”數(shù)據(jù)中分離分析，我們也獲得了很多收益。 每個模型都是單線程的（用于編寫），但是多個模型并行使用事件。 最后，可以將解決方案修改為使用分片或某種fork-join處理。

另一個有趣的觀點是從頭開始重新啟動投影 。

一個好的解決方案是類似于kappa體系結(jié)構(gòu) ：

• 保持過時的預(yù)測正常運行并回答所有查詢。
• 開始一個新的投影，例如到另一個數(shù)據(jù)庫。 只要讓它處理事件，就不要指向任何流量。
• 當(dāng)新的預(yù)測趕上時，請重定向流量并關(guān)閉舊的預(yù)測。

在很小的情況下，尤其是對于開發(fā)而言，甚至可以在同一情況下在線重新啟動。 它取決于以下問題的答案：重新處理所有事件需要多長時間？ 這個投影陳舊30分鐘是否可以接受？ 我們可以在沒有人使用該系統(tǒng)的夜晚或周末進行部署嗎？ 我們需要重播所有歷史嗎？

這里要考慮的另一個因素是持久性。 如果瓶頸太大，無法進一步優(yōu)化，請考慮使用內(nèi)存視圖模型。

加起來

本質(zhì)上，這就是實現(xiàn)使用事件存儲區(qū)的讀取模型所需的全部工作。 有了線性事件存儲并在單個線程中處理所有內(nèi)容，它變得非常簡單。 如此之多，最后實際上只是一個循環(huán)，實現(xiàn)了開頭所示的減少。

在以后的文章中，我將更深入地探討實施預(yù)測的實際問題。

翻譯自: https://www.javacodegeeks.com/2015/09/writing-an-event-sourced-cqrs-read-model.html

探索cqrs和事件源

總結(jié)

以上是生活随笔為你收集整理的探索cqrs和事件源_编写基于事件的CQRS读取模型的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。