一、前言：從物流詳情開始

大家對物流跟蹤都不陌生，它詳細(xì)記錄了在什么時間發(fā)生了什么，并且數(shù)據(jù)作為重要憑證是不可變的。我理解其背后的價值有這么幾個方面：業(yè)務(wù)方可以管控每個子過程、知道目前所處的環(huán)節(jié)；另一方面，當(dāng)需要追溯時候僅僅通過每一步的記錄就可以回放整個歷史過程。

我在之前的文章中提出過“軟件項目也是人類社會生產(chǎn)關(guān)系的范疇，只不過我們所創(chuàng)造的勞動成果看不見摸不著而已”。所以我們可以借鑒物流跟蹤的思路來開發(fā)軟件項目，把復(fù)雜過程拆解為一個個步驟、子過程、狀態(tài)，這和我們事件劃分是一致的，這就是事件驅(qū)動的典型案例。

二、領(lǐng)域事件

領(lǐng)域事件（Domain Events）是領(lǐng)域驅(qū)動設(shè)計（Domain Driven Design，DDD）中的一個概念，用于捕獲我們所建模的領(lǐng)域中所發(fā)生過的事情。

領(lǐng)域事件本身也作為通用語言（Ubiquitous Language）的一部分成為包括領(lǐng)域?qū)＜以趦?nèi)的所有項目成員的交流用語。

比如在前述的跨境物流例子中，貨品達到保稅倉以后需要分派工作人員進行分揀分包，那么“貨品已到達保稅倉”便是一個領(lǐng)域事件。

首先，從業(yè)務(wù)邏輯來說該事件關(guān)系到整個流程的成功或者失敗；同時又將觸發(fā)后續(xù)子流程；而對于業(yè)務(wù)方來說，該事件也是一個標(biāo)志性的里程碑，代表自己的貨品就快配送到自己手中。

所以通常來說，一個領(lǐng)域事件具有以下幾個特征：較高的業(yè)務(wù)價值，有助于形成完整的業(yè)務(wù)閉環(huán)，將導(dǎo)致進一步的業(yè)務(wù)操作。這里還要強調(diào)一點，領(lǐng)域事件具有明確的邊界。

比如：如果你建模的是餐廳的結(jié)賬系統(tǒng)，那么此時的“客戶已到達”便不是你關(guān)心的重點，因為你不可能在客戶到達時就立即向?qū)Ψ揭X，而“客戶已下單”才是對結(jié)賬系統(tǒng)有用的事件。

2.1、建模領(lǐng)域事件

在建模領(lǐng)域事件時，我們應(yīng)該根據(jù)限界上下文中的通用語言來命名事件及屬性。如果事件由聚合上的命令操作產(chǎn)生，那么我們通常根據(jù)該操作方法的名字來命名領(lǐng)域事件。

對于上面的例子“貨品已到達保稅倉”，我們將發(fā)布與之對應(yīng)的領(lǐng)域事件

GoodsArrivedBondedWarehouseEvent（當(dāng)然在明確的界限上下文中也可以去掉聚合的名字，直接建模為ArrivedBondedWarehouseEvent，這都是命名方面的習(xí)慣）。

事件的名字表明了聚合上的命令方法在執(zhí)行成功之后所發(fā)生的事情，換句話說待定項以及不確定的狀態(tài)是不能作為領(lǐng)域事件的。

一個行之有效的方法是畫出當(dāng)前業(yè)務(wù)的狀態(tài)流轉(zhuǎn)圖，包含前置操作以及引起的狀態(tài)變更，這里表達的是已經(jīng)變更完成的狀態(tài)所以我們不用過去時態(tài)表示，比如刪除或者取消，即代表已經(jīng)刪除或者已經(jīng)取消。

然后對于其中的節(jié)點進行事件建模。如下圖是文件云端存儲的業(yè)務(wù)，我們分別對預(yù)上傳、上傳完成確認(rèn)、刪除等環(huán)節(jié)建模“過去時”事件，PreUploadedEvent、ConfirmUploadedEvent、RemovedEvent。

2.2、領(lǐng)域事件代碼解讀

package domain.event;

import java.util.Date;
import java.util.UUID;

public class DomainEvent {

    /**
     * 領(lǐng)域事件還包含了唯一ID，
     * 但是該ID并不是實體（Entity）層面的ID概念，
     * 而是主要用于事件追溯和日志。
     * 如果是數(shù)據(jù)庫存儲，該字段通常為唯一索引。
     */
    private final String id;

    /**
     * 創(chuàng)建時間用于追溯，另一方面不管使用了
     * 哪種事件存儲都有可能遇到事件延遲，
     * 我們通過創(chuàng)建時間能夠確保其發(fā)生順序。
     */
    private final Date occurredOn;

    public DomainEvent() {
        this.id = String.valueOf(UUID.randomUUID());
        this.occurredOn = new Date();
    }
}

在創(chuàng)建領(lǐng)域事件時，需要注意2點：

• 領(lǐng)域事件本身應(yīng)該是不變的(Immutable)；
• 領(lǐng)域事件應(yīng)該攜帶與事件發(fā)生時相關(guān)的上下文數(shù)據(jù)信息，但是并不是整個聚合根的狀態(tài)數(shù)據(jù)。例如，在創(chuàng)建訂單時可以攜帶訂單的基本信息，而對于用戶更新訂單收貨地址事件AddressUpdatedEvent事件，只需要包含訂單、用戶以及新的地址等信息即可。

public class AddressUpdatedEvent extends DomainEvent {
    //通過userId+orderId來校驗訂單的合法性；
    private String userId; 
    private String orderId;
    //新的地址
    private Address address;
    //略去具體業(yè)務(wù)邏輯
}

2.3、領(lǐng)域事件的存儲

事件的不可變性與可追溯性都決定了其必須要持久化的原則，我們來看看常見的幾種方案。

2.3.1、單獨的EventStore

有的業(yè)務(wù)場景中會創(chuàng)建一個單獨的事件存儲中心，可能是Mysql、Redis、Mongo、甚至文件存儲等。這里以Mysql舉例，business_code、event_code用來區(qū)分不同業(yè)務(wù)的不同事件，具體的命名規(guī)則可以根據(jù)實際需要。

這里需要注意該數(shù)據(jù)源與業(yè)務(wù)數(shù)據(jù)源不一致的場景，我們要確保當(dāng)業(yè)務(wù)數(shù)據(jù)更新以后事件能夠準(zhǔn)確無誤的記錄下來，實踐中盡量避免使用分布式事務(wù)，或者盡量避免其跨庫的場景，否則你就得想想如何補償了。千萬要避免，用戶更新了收貨地址，但是AddressUpdatedEvent事件保存失敗。

總的原則就是對分布式事務(wù)Say No，無論如何，我相信方法總比問題多，在實踐中我們總可以想到解決方案，區(qū)別在于該方案是否簡潔、是否做到了解耦。

# 考慮是否需要分表，事件存儲建議邏輯簡單
CREATE TABLE `event_store` (
  `event_id` int(11) NOT NULL auto increment,
  `event_code` varchar(32) NOT NULL,
  `event_name` varchar(64) NOT NULL,
  `event_body` varchar(4096) NOT NULL,
  `occurred_on` datetime NOT NULL，
  `business_code` varchar(128) NOT NULL,
  UNIQUE KEY (`event id`)
) ENGINE=InnoDB COMMENT '事件存儲表';

2.3.2、與業(yè)務(wù)數(shù)據(jù)一起存儲

在分布式架構(gòu)中，每個模塊都做的相對比較小，準(zhǔn)確的說是“自治”。如果當(dāng)前業(yè)務(wù)數(shù)據(jù)量較小，可以將事件與業(yè)務(wù)數(shù)據(jù)一起存儲，用相關(guān)標(biāo)識區(qū)分是真實的業(yè)務(wù)數(shù)據(jù)還是事件記錄；或者在當(dāng)前業(yè)務(wù)數(shù)據(jù)庫中建立該業(yè)務(wù)自己的事件存儲，但是要考慮到事件存儲的量級必然大于真實的業(yè)務(wù)數(shù)據(jù)，考慮是否需要分表。

這種方案的優(yōu)勢：數(shù)據(jù)自治；避免分布式事務(wù)；不需要額外的事件存儲中心。當(dāng)然其劣勢就是不能復(fù)用。

2.4、領(lǐng)域事件如何發(fā)布

2.4.1、由領(lǐng)域聚合發(fā)送領(lǐng)域事件

/*
* 一個關(guān)于比賽的充血模型例子
* 貧血模型會構(gòu)造一個MatchService，我們這里通過模型來觸發(fā)相應(yīng)的事件
* 本例中略去了具體的業(yè)務(wù)細(xì)節(jié)
*/
public class Match {
    public void start() {
        //構(gòu)造Event....
        MatchEvent matchStartedEvent = new MatchStartedEvent();
        //略去具體業(yè)務(wù)邏輯
        DefaultDomainEventBus.publish(matchStartedEvent);
    }

    public void finish() {
        //構(gòu)造Event....
        MatchEvent matchFinishedEvent = new MatchFinishedEvent();
        //略去具體業(yè)務(wù)邏輯
        DefaultDomainEventBus.publish(matchFinishedEvent);
    }

    //略去Match對象基本屬性
}

2.4.2、事件總線VS消息中間件

微服務(wù)內(nèi)的領(lǐng)域事件可以通過事件總線或利用應(yīng)用服務(wù)實現(xiàn)不同聚合之間的業(yè)務(wù)協(xié)同。即微服務(wù)內(nèi)發(fā)生領(lǐng)域事件時，由于大部分事件的集成發(fā)生在同一個線程內(nèi)，不一定需要引入消息中間件。但一個事件如果同時更新多個聚合數(shù)據(jù)，按照 DDD“一個事務(wù)只更新一個聚合根”的原則，可以考慮引入消息中間件，通過異步化的方式，對微服務(wù)內(nèi)不同的聚合根采用不同的事務(wù)

三、Saga分布式事務(wù)

3.1、Saga概要

我們看看如何使用 Saga 模式維護數(shù)據(jù)一致性？

Saga 是一種在微服務(wù)架構(gòu)中維護數(shù)據(jù)一致性的機制，它可以避免分布式事務(wù)所帶來的問題。

一個 Saga 表示需要更新的多個服務(wù)中的一個，即Saga由一連串的本地事務(wù)組成。每一個本地事務(wù)負(fù)責(zé)更新它所在服務(wù)的私有數(shù)據(jù)庫，這些操作仍舊依賴于我們所熟悉的ACID事務(wù)框架和函數(shù)庫。

模式：Saga

通過使用異步消息來協(xié)調(diào)一系列本地事務(wù)，從而維護多個服務(wù)之間的數(shù)據(jù)一致性。

請參閱（強烈建議）：https://microservices.io/patterns/data/saga.html

Saga與TCC相比少了一步Try的操作，TCC無論最終事務(wù)成功失敗都需要與事務(wù)參與方交互兩次。而Saga在事務(wù)成功的情況下只需要與事務(wù)參與方交互一次, 如果事務(wù)失敗，需要額外進行補償回滾。

• 每個Saga由一系列sub-transaction Ti 組成；
• 每個Ti 都有對應(yīng)的補償動作Ci，補償動作用于撤銷Ti造成的結(jié)果；

可以看到，和TCC相比，Saga沒有“預(yù)留”動作，它的Ti就是直接提交到庫。

Saga的執(zhí)行順序有兩種：

• success：T1, T2, T3, ..., Tn ；
• failure：T1, T2, ..., Tj, Cj,..., C2, C1，其中0 < j < n；

所以我們可以看到Saga的撤銷十分關(guān)鍵，可以說使用Saga的難點就在于如何設(shè)計你的回滾策略。

3.2、Saga實現(xiàn)

通過上面的例子我們對Saga有了初步的體感，現(xiàn)在來深入探討下如何實現(xiàn)。當(dāng)通過系統(tǒng)命令啟動Saga時，協(xié)調(diào)邏輯必須選擇并通知第一個Saga參與方執(zhí)行本地事務(wù)。一旦該事務(wù)完成，Saga協(xié)調(diào)選擇并調(diào)用下一個Saga參與方。

這個過程一直持續(xù)到Saga執(zhí)行完所有步驟。如果任何本地事務(wù)失敗，則 Saga必須以相反的順序執(zhí)行補償事務(wù)。以下幾種不同的方法可用來構(gòu)建Saga的協(xié)調(diào)邏輯。

3.2.1、協(xié)同式（choreography）

把 Saga 的決策和執(zhí)行順序邏輯分布在 Saga的每一個參與方中，它們通過交換事件的方式來進行溝通。

（引用于《微服務(wù)架構(gòu)設(shè)計模式》相關(guān)章節(jié)）

Order服務(wù)創(chuàng)建一個Order并發(fā)布OrderCreated事件。

Consumer服務(wù)消費OrderCreated事件，驗證消費者是否可以下訂單，并發(fā)布ConsumerVerified事件。

Kitchen服務(wù)消費OrderCreated事件，驗證訂單，在CREATE_PENDING狀態(tài)下創(chuàng)建故障單，并發(fā)布TicketCreated事件。

Accounting服務(wù)消費OrderCreated事件并創(chuàng)建一個處于PENDING狀態(tài)的Credit CardAuthorization。

Accounting服務(wù)消費TicketCreated和ConsumerVerified事件，向消費者的信用卡收費，并發(fā)布信用卡授權(quán)失敗事件。

Kitchen服務(wù)使用信用卡授權(quán)失敗事件并將故障單的狀態(tài)更改為REJECTED。

訂單服務(wù)消費信用卡授權(quán)失敗事件，并將訂單狀態(tài)更改為已拒絕。

3.2.2、編排式（orchestration）

把Saga的決策和執(zhí)行順序邏輯集中在一個Saga編排器類中。Saga 編排器發(fā)出命令式消息給各個 Saga 參與方，指示這些參與方服務(wù)完成具體操作（本地事務(wù)）。類似于一個狀態(tài)機，當(dāng)參與方服務(wù)完成操作以后會給編排器發(fā)送一個狀態(tài)指令，以決定下一步做什么。

（引用于《微服務(wù)架構(gòu)設(shè)計模式》相關(guān)章節(jié)）

我們來分析一下執(zhí)行流程

Order Service首先創(chuàng)建一個Order和一個創(chuàng)建訂單控制器。之后，路徑的流程如下：

Saga orchestrator向Consumer Service發(fā)送Verify Consumer命令。

Consumer Service回復(fù)Consumer Verified消息。

Saga orchestrator向Kitchen Service發(fā)送Create Ticket命令。

Kitchen Service回復(fù)Ticket Created消息。

Saga協(xié)調(diào)器向Accounting Service發(fā)送授權(quán)卡消息。

Accounting服務(wù)部門使用卡片授權(quán)消息回復(fù)。

Saga orchestrator向Kitchen Service發(fā)送Approve Ticket命令。

Saga orchestrator向訂單服務(wù)發(fā)送批準(zhǔn)訂單命令。

3.2.3、補償策略

之前的描述中我們說過Saga最重要的是如何處理異常，狀態(tài)機還定義了許多異常狀態(tài)。如上面的6就會發(fā)生失敗，觸發(fā)AuthorizeCardFailure，此時我們就要結(jié)束訂單并把之前提交的事務(wù)進行回滾。這里面要區(qū)分哪些是校驗性事務(wù)、哪些是需要補償?shù)氖聞?wù)。

一個Saga由三種不同類型的事務(wù)組成：可補償性事務(wù)（可以回滾，因此有一個補償事務(wù))；關(guān)鍵性事務(wù)（這是 Saga的成敗關(guān)鍵點，比如4賬戶代扣）；以及可重復(fù)性事務(wù)，它不需要回滾并保證能夠完成（比如6更新狀態(tài)）。

在Create Order Saga 中，createOrder()、createTicket()步驟是可補償性事務(wù)且具有撤銷其更新的補償事務(wù)。

verifyConsumerDetails()事務(wù)是只讀的，因此不需要補償事務(wù)。authorizeCreditCard()事務(wù)是這個 Saga的關(guān)鍵性事務(wù)。如果消費者的信用卡可以授權(quán)，那么這個Saga保證完成。approveTicket()和approveRestaurantOrder()步驟是在關(guān)鍵性事務(wù)之后的可重復(fù)性事務(wù)。

認(rèn)真拆解每個步驟、然后評估其補償策略尤為重要，正如你看到的，每種類型的事務(wù)在對策中扮演著不同的角色。

四、CQRS

前面講述了事件的概念，又分析了Saga如何解決復(fù)雜事務(wù)，現(xiàn)在我們來看看CQRS為什么在DDD中廣泛被采用。除了讀寫分離的特征以外，我們用事件驅(qū)動的方式來實踐Command邏輯能有效降低業(yè)務(wù)的復(fù)雜度。

當(dāng)你明白如何建模事件、如何規(guī)避復(fù)雜事務(wù)，明白什么時候用消息中間件、什么時候采用事件總線，才能理解為什么是CQRS、怎么正確應(yīng)用。

下面是我們項目中的設(shè)計，這里為什么會出現(xiàn)Read/Write Service，是為了封裝調(diào)用，service內(nèi)部是基于聚合發(fā)送事件。因為我發(fā)現(xiàn)在實際項目中，很多人都會第一時間問我要XXXService而不是XXX模型，所以在DDD沒有完全普及的項目中建議大家采取這種居中策略。這也符合咱們的解耦，對方依賴我的抽象能力，然而我內(nèi)部是基于DDD還是傳統(tǒng)的流程代碼對其是無關(guān)透明的。

我們先來看看事件以及處理器的時序關(guān)系。

這里還是以文件云端存儲業(yè)務(wù)為例，下面是一些處理器的核心代碼。注釋行是對代碼功能、用法以及擴展方面的解讀，請認(rèn)真閱讀。

package domain;

import domain.event.DomainEvent;
import domain.handler.event.DomainEventHandler;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class DomainRegistry {

    private Map<String, List<DomainEventHandler>> handlerMap =
        new HashMap<String, List<DomainEventHandler>>();

    private static DomainRegistry instance;

    private DomainRegistry() {
    }

    public static DomainRegistry getInstance() {
        if (instance == null) {
            instance = new DomainRegistry();
        }
        return instance;
    }

    public Map<String, List<DomainEventHandler>> getHandlerMap() {
        return handlerMap;
    }

    public List<DomainEventHandler> find(String name) {
        if (name == null) {
            return null;
        }
        return handlerMap.get(name);
    }

    //事件注冊與維護，register分多少個場景根據(jù)業(yè)務(wù)拆分，
    //這里是業(yè)務(wù)流的核心。如果多個事件需要維護前后依賴關(guān)系，
    //可以維護一個priority邏輯
    public void register(Class<? extends DomainEvent> domainEvent,
                         DomainEventHandler handler) {
        if (domainEvent == null) {
            return;
        }
        if (handlerMap.get(domainEvent.getName()) == null) {
            handlerMap.put(domainEvent.getName(), new ArrayList<DomainEventHandler>());
        }
        handlerMap.get(domainEvent.getName()).add(handler);
        //按照優(yōu)先級進行事件處理器排序
        。。。
    }
}

文件上傳完畢事件的例子。

package domain.handler.event;

import domain.DomainRegistry;
import domain.StateDispatcher;
import domain.entity.meta.MetaActionEnums;
import domain.event.DomainEvent;
import domain.event.MetaEvent;
import domain.repository.meta.MetaRepository;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * @Description:一個事件操作的處理器
 * 我們混合使用了Saga的兩種模式，外層事件交互；
 * 對于單個復(fù)雜的事件內(nèi)部采取狀態(tài)流轉(zhuǎn)實現(xiàn)。
 */

@Component
public class MetaConfirmUploadedHandler implements DomainEventHandler {

    @Resource
    private MetaRepository metaRepository;

    public void handle(DomainEvent event) {
        //1.我們在當(dāng)前的上下文中定義個ThreadLocal變量
        //用于存放事件影響的聚合根信息(線程共享)

        //2.當(dāng)然如果有需要額外的信息，可以基于event所
        //攜帶的信息構(gòu)造Specification從repository獲取
        // 代碼示例
        // metaRepository.queryBySpecification(SpecificationFactory.build(event));

        DomainEvent domainEvent = metaRepository.load();

        //此處是我們的邏輯
        。。。。

        //對于單個操作比較復(fù)雜的，可以使用狀態(tài)流轉(zhuǎn)進一步拆分
        domainEvent.setStatus(nextState);
        //在事件觸發(fā)之后，仍需要一個狀態(tài)跟蹤器來解決大事務(wù)問題
        //Saga編排式
        StateDispatcher.dispatch();
    }

    @PostConstruct
    public void autoRegister() {
        //此處可以更加細(xì)分，注冊在哪一類場景中，這也是事件驅(qū)動的強大、靈活之處。
        //避免了if...else判斷。我們可以有這樣的意識，一旦你的邏輯里面充斥了大量
        //switch、if的時候來看看自己注冊的場景是否可以繼續(xù)細(xì)分
        DomainRegistry.getInstance().register(MetaEvent.class, this);
    }

    public String getAction() {
        return MetaActionEnums.CONFIRM_UPLOADED.name();
    }

    //適用于前后依賴的事件，通過優(yōu)先級指定執(zhí)行順序
    public Integer getPriority() {
        return PriorityEnums.FIRST.getValue();
    }
}

事件總線邏輯

package domain;

import domain.event.DomainEvent;
import domain.handler.event.DomainEventHandler;
import java.util.List;


public class DefaultDomainEventBus {

    public static void publish(DomainEvent event, String action,
                               EventCallback callback) {

        List<DomainEventHandler> handlers = DomainRegistry.getInstance().
            find(event.getClass().getName());
        handlers.stream().forEach(handler -> {
            if (action != null && action.equals(handler.getAction())) {
                Exception e = null;
                boolean result = true;
                try {
                    handler.handle(event);
                } catch (Exception ex) {
                    e = ex;
                    result = false;
                    //自定義異常處理
                    。。。
                } finally {
                    //write into event store
                    saveEvent(event);
                }

                //根據(jù)實際業(yè)務(wù)處理回調(diào)場景，DefaultEventCallback可以返回
                if (callback != null) {
                    callback.callback(event, action, result, e);       
                }
            }
        });
    }
}

五、自治服務(wù)和系統(tǒng)

DDD中強調(diào)限界上下文的自治特性，事實上，從更小的粒度來看，對象仍然需要具備自治的這四個特性，即：最小完備、自我履行、穩(wěn)定空間、獨立進化。其中自我履行是重點，因為不強依賴外部所以穩(wěn)定、因為穩(wěn)定才可能獨立進化。這就是六邊形架構(gòu)在DDD中較為普遍的原因。

六、結(jié)語

本文所講述的事件、Saga、CQRS的方案均可以單獨使用，可以應(yīng)用到你的某個method、或者你的整個package。項目中我們并不一定要實踐一整套CQRS，只要其中的某些思想解決了我們項目中的某個問題就足夠了。

也許你現(xiàn)在已經(jīng)磨刀霍霍，準(zhǔn)備在項目中實踐一下這些技巧。不過我們要明白“每一個硬幣都有兩面性”，我們不僅看到高擴展、解耦的、易編排的優(yōu)點以外，仍然要明白其所帶來的問題。利弊分析以后再去決定如何實現(xiàn)才是正確的應(yīng)對之道。

• 這類編程模式有一定的學(xué)習(xí)曲線；
• 基于消息傳遞的應(yīng)用程序的復(fù)雜性；
• 處理事件的演化有一定難度；
• 刪除數(shù)據(jù)存在一定難度；
• 查詢事件存儲庫非常有挑戰(zhàn)性。

不過我們還是要認(rèn)識到在其適合的場景中，六邊形架構(gòu)以及DDD戰(zhàn)術(shù)將加速我們的領(lǐng)域建模過程，也迫使我們從嚴(yán)格的通用語言角度來解釋一個領(lǐng)域，而不是一個個需求。任何更強調(diào)核心域而不是技術(shù)實現(xiàn)的方式都可以增加業(yè)務(wù)價值，并使我們獲得更大的競爭優(yōu)勢。

以上就是詳解領(lǐng)域驅(qū)動設(shè)計之事件驅(qū)動與CQRS的詳細(xì)內(nèi)容，更多關(guān)于領(lǐng)域驅(qū)動設(shè)計 事件驅(qū)動與CQRS的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論