领域驱动设计 (DDD) 实践之路（二）：事件驱动与 CQRS

package domain.event;

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

/**
 * @Description:
 * @Author: zhangwenbo
 * @Since: 2019/3/6
 */
public class DomainEvent {

    /**
     * 领域事件还包含了唯一ID，
     * 但是该ID并不是实体（Entity）层面的ID概念，
     * 而是主要用于事件追溯和日志。
     * 如果是数据库存储，该字段通常为唯一索引。
     */
    private final String id;

    /**
     * 创建时间用于追溯，另一方面不管使用了
     * 哪种事件存储都有可能遇到事件延迟，
     * 我们通过创建时间能够确保其发生顺序。
     */
    private final Date occurredOn;

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

（滑动可查看）

在创建领域事件时，需要注意2点：

• 领域事件本身应该是不变的(Immutable)；

• 领域事件应该携带与事件发生时相关的上下文数据信息，但是并不是整个聚合根的状态数据。例如，在创建订单时可以携带订单的基本信息，而对于用户更新订单收货地址事件AddressUpdatedEvent事件，只需要包含订单、用户以及新的地址等信息即可。

public class AddressUpdatedEvent extends DomainEvent {
    //通过userId+orderId来校验订单的合法性；
    private String userId; 
    private String orderId;
    //新的地址
    private Address address;
    //略去具体业务逻辑
}

（滑动可查看）

事件的不可变性与可追溯性都决定了其必须要持久化的原则，我们来看看常见的几种方案。

3.1单独的EventStore

有的业务场景中会创建一个单独的事件存储中心，可能是 Mysql 、 Redis 、Mongo、甚至文件存储等。这里以Mysql举例，business_code、event_code用来区分不同业务的不同事件，具体的命名规则可以根据实际需要。

这里需要注意该数据源与业务数据源不一致的场景，我们要确保当业务数据更新以后事件能够准确无误的记录下来，实践中尽量避免使用分布式事务，或者尽量避免其跨库的场景，否则你就得想想如何补偿了。千万要避免，用户更新了收货地址，但是AddressUpdatedEvent事件保存失败。

总的原则就是对分布式事务Say No，无论如何，我相信方法总比问题多，在实践中我们总可以想到解决方案，区别在于该方案是否简洁、是否做到了解耦。

# 考虑是否需要分表，事件存储建议逻辑简单
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 '事件存储表';

（滑动可查看）

3.2 与业务数据一起存储

在分布式架构中，每个模块都做的相对比较小，准确的说是“自治”。如果当前业务数据量较小，可以将事件与业务数据一起存储，用相关标识区分是真实的业务数据还是事件记录；或者在当前业务数据库中建立该业务自己的事件存储，但是要考虑到事件存储的量级必然大于真实的业务数据，考虑是否需要分表。

这种方案的优势：数据自治；避免分布式事务；不需要额外的事件存储中心。当然其劣势就是不能复用。

/*
* 一个关于比赛的充血模型例子
* 贫血模型会构造一个MatchService，我们这里通过模型来触发相应的事件
* 本例中略去了具体的业务细节
*/
public class Match {
    public void start() {
        //构造Event....
        MatchEvent matchStartedEvent = new MatchStartedEvent();
        //略去具体业务逻辑
        DefaultDomainEventBus.publish(matchStartedEvent);
    }

    public void finish() {
        //构造Event....
        MatchEvent matchFinishedEvent = new MatchFinishedEvent();
        //略去具体业务逻辑
        DefaultDomainEventBus.publish(matchFinishedEvent);
    }

    //略去Match对象基本属性
}

（滑动可查看）

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;

/**
 * @Description: 事件注册逻辑
 * @Author: zhangwenbo
 * @Since: 2019/3/6
 */

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分多少个场景根据业务拆分，
    //这里是业务流的核心。如果多个事件需要维护前后依赖关系，
    //可以维护一个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);
        //按照优先级进行事件处理器排序
        。。。
    }
}

（滑动可查看）

文件上传完毕事件的例子。

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的两种模式，外层事件交互；
 * 对于单个复杂的事件内部采取状态流转实现。
 * @Author: zhangwenbo
 * @Since: 2019/3/6
 */

@Component
public class MetaConfirmUploadedHandler implements DomainEventHandler {

    @Resource
    private MetaRepository metaRepository;

    public void handle(DomainEvent event) {
        //1.我们在当前的上下文中定义个ThreadLocal变量
        //用于存放事件影响的聚合根信息(线程共享)

        //2.当然如果有需要额外的信息，可以基于event所
        //携带的信息构造Specification从repository获取
        // 代码示例
        // metaRepository.queryBySpecification(SpecificationFactory.build(event));

        DomainEvent domainEvent = metaRepository.load();

        //此处是我们的逻辑
        。。。。

        //对于单个操作比较复杂的，可以使用状态流转进一步拆分
        domainEvent.setStatus(nextState);
        //在事件触发之后，仍需要一个状态跟踪器来解决大事务问题
        //Saga编排式
        StateDispatcher.dispatch();
    }

    @PostConstruct
    public void autoRegister() {
        //此处可以更加细分，注册在哪一类场景中，这也是事件驱动的强大、灵活之处。
        //避免了if...else判断。我们可以有这样的意识，一旦你的逻辑里面充斥了大量
        //switch、if的时候来看看自己注册的场景是否可以继续细分
        DomainRegistry.getInstance().register(MetaEvent.class, this);
    }

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

    //适用于前后依赖的事件，通过优先级指定执行顺序
    public Integer getPriority() {
        return PriorityEnums.FIRST.getValue();
    }
}

（滑动可查看）

事件总线逻辑

package domain;

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

/**
 * @Description:
 * @Author: zhangwenbo
 * @Since: 2019/3/6
 */

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);
                }

                //根据实际业务处理回调场景，DefaultEventCallback可以返回
                if (callback != null) {
                    callback.callback(event, action, result, e);       
                }
            }
        });
    }
}

（滑动可查看）

不过我们还是要认识到在其适合的场景中，六边形架构以及DDD战术将加速我们的领域建模过程，也迫使我们从严格的通用语言角度来解释一个领域，而不是一个个需求。任何更强调核心域而不是技术实现的方式都可以增加业务价值，并使我们获得更大的竞争优势。

附：参考文献

1. Pattern: Saga

2. 分布式事务：Saga模式

3. 书籍：《微服务架构设计模式》

• 领域驱动设计(DDD)实践之路(一)

• vivo 大规模特征存储实践

• 分布式定时任务调度框架实践

• 揭秘 vivo 如何打造千万级 DAU 活动中台 - 启航篇

vivo移动互联网是基于vivo 智能手机所建立的完整移动互联网生态圈，围绕vivo大数据运营，打造包括应用、游戏、资讯、品牌、电商、内容、金融、搜索的全方位服务生态，满足海量用户的多样化需求。