领域驱动设计战术模式--领域事件

时间 2019-12-07

标签领域驱动设计战术模式事件繁體版

原文原文链接

使用领域事件来捕获发生在领域中的一些事情。git

领域驱动实践者发现他们能够经过了解更多发生在问题域中的事件，来更好的理解问题域。这些事件，就是领域事件，主要是与领域专家一块儿进行知识提炼环节中得到。redis

领域事件，能够用于一个限界上下文内的领域模型，也可使用消息队列在限界上下文间进行异步通讯。spring

1 理解领域事件

领域事件是领域专家所关心的发生在领域中的一些事件。

将领域中所发生的活动建模成一系列离散事件。每一个事件都用领域对象表示。领域事件是领域模型的组成部分，表示领域中所发生的事情。数据库

领域事件的主要用途：设计模式

保证聚合间的数据一致性
替换批量处理
实现事件源模式
进行限界上下文集成

2 实现领域事件

领域事件表示已经发生的某种事实，该事实在发生后便不会改变。所以，领域事件一般建模成值对象。
但，这也有特殊的状况，为了迎合序列化和反序列化框架需求，在建模时，常常会进行必定的妥协。缓存

2.1 建立领域事件

2.1.1 事件命名

在建模领域事件时，咱们应该根据限界上下文中的通用语言来命名事件。

若是事件由聚合上的命令操做产生，一般根据该操做方法的名字来命名事件。事件名字代表聚合上的命令方法在执行成功后的事实。即事件命名须要反映过去发生过的事情。架构

public class AccountEnabledEvent extends AbstractAggregateEvent<Long, Account> {
    public AccountEnabledEvent(Account source) {
        super(source);
    }
}

2.1.2 事件属性

事件的属性主要用于驱动后续业务流程。固然，也会拥有一些通用属性。

事件具备一些通用属性，如：app

惟一标识
occurredOn 发生时间
type 事件类型
source 事件发生源（只针对由聚合产生的事件）

通用属性可使用事件接口来规范。框架

接口或类	含义
DomainEvent	通用领域事件接口
AggregateEvent	由聚合发布的通用领域事件接口
AbstractDomainEvent	DomainEvent 实现类，维护 id 和建立时间
AbstractAggregateEvent	AggregateEvent 实现类，继承子 AbstractDomainEvent，并添加 source 属性

但，事件最主要的仍是业务属性。咱们须要考虑，是谁致使事件的发生，这可能涉及产生事件的聚合或其余参与该操做的聚合，也多是其余任何类型的操做数据。

2.1.3 事件方法

事件是事实的描述，自己不会有太多的业务操做。

领域事件一般被设计为不变对象，事件所携带的数据已经反映出该事件的来源。事件构造函数完成状态初始化，同时提供属性的 getter 方法。dom

2.1.4 事件惟一标识

这里须要注意的是事件惟一标识，一般状况下，事件是不可变的，那为何会涉及惟一标识的概念呢？

对于从聚合中发布出来的领域事件，使用事件的名称、产生事件的标识、事件发生的时间等足以对不一样的事件进行区分。但，这样会增长事件比较的复杂性。

对于由调用方发布的事件，咱们将领域事件建模成聚合，能够直接使用聚合的惟一标识做为事件的标识。

事件惟一标识的引入，会大大减小事件比较的复杂性。但，其最大的意义在于限界上下文的集成。

当咱们须要将领域事件发布到外部的限界上下文时，惟一标识就是一种必然。为了保证事件投递的幂等性，在发送端，咱们可能会进行屡次发送尝试，直至明确发送成功为止；而在接收端，当接收到事件后，须要对事件进行重复性检测，以保障事件处理的幂等性。此时，事件的惟一标识即可以做为事件去重的依据。

事件惟一标识，自己对领域建模影响不大，但对技术处理好处巨大。所以，将它做为通用属性进行管理。

2.2 发布领域事件

咱们如何避免领域事件与处理者间的耦合呢？

一种简单高效的方式即是使用观察者模式，这种模式能够在领域事件和外部组件间进行解耦。

2.2.1 发布订阅模型

为了统一，咱们须要定义了一套接口和实现类，以基于观察者模式，完成事件的发布。

涉及接口和实现类以下：

接口或类	含义
DomainEventPublisher	用于发布领域事件
DomainEventHandlerRegistry	用于注册 DomainEventHandler
DomainEventBus	扩展自 DomainEventPublisher 和 DomainEventHandlerRegistry 用于发布和管理领域事件处理器
DefaultDomainEventBus	DomainEventBus 默认实现
DomainEventHandler	用于处理领域事件
DomainEventSubscriber	用于判断是否接受领域事件
DomainEventExecutor	用于执行领域事件处理器

使用实例如 DomainEventBusTest 所示：

public class DomainEventBusTest {
    private DomainEventBus domainEventBus;

    @Before
    public void setUp() throws Exception {
        this.domainEventBus = new DefaultDomainEventBus();
    }

    @After
    public void tearDown() throws Exception {
        this.domainEventBus = null;
    }

    @Test
    public void publishTest(){
        // 建立事件处理器
        TestEventHandler eventHandler = new TestEventHandler();
        // 注册事件处理器
        this.domainEventBus.register(TestEvent.class, eventHandler);

        // 发布事件
        this.domainEventBus.publish(new TestEvent("123"));

        // 检测事件处理器是够运行
        Assert.assertEquals("123", eventHandler.data);
    }

    @Value
    class TestEvent extends AbstractDomainEvent{
        private String data;
    }

    class TestEventHandler implements DomainEventHandler<TestEvent>{
        private String data;
        @Override
        public void handle(TestEvent event) {
            this.data = event.getData();
        }
    }
}

在构建完发布订阅结构后，须要将其与领域模型进行关联。领域模型如何获取 Publisher，事件处理器如何进行订阅。

2.2.2 基于 ThreadLocal 的事件发布

比较经常使用的方案即是将 DomainEventBus 绑定到线程上下文。这样，只要是同一调用线程均可以方便的获取 DomainEventBus 对象。

具体的交互以下：

DomainEventBusHolder 用于管理 DomainEventBus。

public class DomainEventBusHolder {
    private static final ThreadLocal<DomainEventBus> THREAD_LOCAL = new ThreadLocal<DomainEventBus>(){
        @Override
        protected DomainEventBus initialValue() {
            return new DefaultDomainEventBus();
        }
    };

    public static DomainEventPublisher getPubliser(){
        return THREAD_LOCAL.get();
    }

    public static DomainEventHandlerRegistry getHandlerRegistry(){
        return THREAD_LOCAL.get();
    }

    public static void clean(){
        THREAD_LOCAL.remove();
    }
}

Account 的 enable 直接使用 DomainEventBusHolder 进行发布。

public class Account extends JpaAggregate {

    public void enable(){
        AccountEnabledEvent event = new AccountEnabledEvent(this);
        DomainEventBusHolder.getPubliser().publish(event);
    }
}

public class AccountEnabledEvent extends AbstractAggregateEvent<Long, Account> {
    public AccountEnabledEvent(Account source) {
        super(source);
    }
}

AccountApplication 完成订阅器注册以及业务方法调用。

public class AccountApplication extends AbstractApplication {
    private static final Logger LOGGER = LoggerFactory.getLogger(AccountApplication.class);

    @Autowired
    private AccountRepository repository;

    public void enable(Long id){
        // 清理以前绑定的 Handler
        DomainEventBusHolder.clean();

        // 注册 EventHandler
        AccountEnableEventHandler enableEventHandler = new AccountEnableEventHandler();
        DomainEventBusHolder.getHandlerRegistry().register(AccountEnabledEvent.class, enableEventHandler);

        Optional<Account> accountOptional = repository.getById(id);
        if (accountOptional.isPresent()) {
            Account account = accountOptional.get();
            // enable 使用 DomainEventBusHolder 直接发布事件
            account.enable();
            repository.save(account);
        }
    }
    
    class AccountEnableEventHandler implements DomainEventHandler<AccountEnabledEvent>{

        @Override
        public void handle(AccountEnabledEvent event) {
            LOGGER.info("handle enable event");
        }
    }
}

2.2.3 基于实体缓存的事件发布

先将事件缓存在实体中，在实体状态成功持久化到存储后，再进行事件发布。

具体交互以下：

实例代码以下：

public class Account extends JpaAggregate {

    public void enable(){
        AccountEnabledEvent event = new AccountEnabledEvent(this);
        registerEvent(event);
    }
}

Account 的 enable 方法，调用 registerEvent 对事件进行注册。

@MappedSuperclass
public abstract class AbstractAggregate<ID> extends AbstractEntity<ID> implements Aggregate<ID> {
    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractAggregate.class);

    @JsonIgnore
    @QueryTransient
    @Transient
    @org.springframework.data.annotation.Transient
    private final transient List<DomainEventItem> events = Lists.newArrayList();

    protected void registerEvent(DomainEvent event) {
        events.add(new DomainEventItem(event));
    }

    protected void registerEvent(Supplier<DomainEvent> eventSupplier) {
        this.events.add(new DomainEventItem(eventSupplier));
    }

    @Override
    @JsonIgnore
    public List<DomainEvent> getEvents() {
        return Collections.unmodifiableList(events.stream()
                .map(eventSupplier -> eventSupplier.getEvent())
                .collect(Collectors.toList()));
    }

    @Override
    public void cleanEvents() {
        events.clear();
    }


    private class DomainEventItem {
        DomainEventItem(DomainEvent event) {
            Preconditions.checkArgument(event != null);
            this.domainEvent = event;
        }

        DomainEventItem(Supplier<DomainEvent> supplier) {
            Preconditions.checkArgument(supplier != null);
            this.domainEventSupplier = supplier;
        }

        private DomainEvent domainEvent;
        private Supplier<DomainEvent> domainEventSupplier;

        public DomainEvent getEvent() {
            if (domainEvent != null) {
                return domainEvent;
            }
            DomainEvent event = this.domainEventSupplier != null ? this.domainEventSupplier.get() : null;
            domainEvent = event;
            return domainEvent;
        }
    }
}

registerEvent 方法在 AbstractAggregate 中，registerEvent 方法将事件保存到 events 集合，getEvents 方法获取全部事件，cleanEvents 方法清理缓存的事件。

Application 实例以下：

@Service
public class AccountApplication extends AbstractApplication {
    private static final Logger LOGGER = LoggerFactory.getLogger(AccountApplication.class);

    @Autowired
    private AccountRepository repository;

    @Autowired
    private DomainEventBus domainEventBus;

    @PostConstruct
    public void init(){
        // 使用 Spring 生命周期注册事件处理器
        this.domainEventBus.register(AccountEnabledEvent.class, new AccountEnableEventHandler());
    }

    public void enable(Long id){
        Optional<Account> accountOptional = repository.getById(id);
        if (accountOptional.isPresent()) {
            Account account = accountOptional.get();
            // enable 将事件缓存在 account 中
            account.enable();
            repository.save(account);
            List<DomainEvent> events = account.getEvents();
            if (!CollectionUtils.isEmpty(events)){
                // 成功持久化后，对事件进行发布
                this.domainEventBus.publishAll(events);
            }
        }
    }

    class AccountEnableEventHandler implements DomainEventHandler<AccountEnabledEvent>{

        @Override
        public void handle(AccountEnabledEvent event) {
            LOGGER.info("handle enable event");
        }
    }
}

AccountApplication 的 init 方法完成事件监听器的注册，enable 方法在实体成功持久化后，将缓存的事件经过 DomainEventBus 实例 publish 出去。

2.2.4 由调用方发布事件

一般状况下，领域事件是由聚合的命令方法产生，并在命令方法执行成功后，进行事件的发布。
有时，领域事件并非聚合中的命令方法产生的，而是由用户所发生的请求产生。

此时，咱们须要将领域事件建模成一个聚合，而且拥有本身的资源库。但，因为领域事件表示的是过去发生的事情，所以资源库只作追加操做，不能对事件进行修改和删除功能。

例如，对用户点击事件进行发布。

@Entity
@Data
public class ClickAction extends JpaAggregate implements DomainEvent {
    @Setter(AccessLevel.PRIVATE)
    private Long userId;
    @Setter(AccessLevel.PRIVATE)
    private String menuId;

    public ClickAction(Long userId, String menuId){
        Preconditions.checkArgument(userId != null);
        Preconditions.checkArgument(StringUtils.isNotEmpty(menuId));

        setUserId(userId);
        setMenuId(menuId);
    }

    @Override
    public String id() {
        return String.valueOf(getId());
    }

    @Override
    public Date occurredOn() {
        return getCreateTime();
    }

}

ClickAction 继承自 JpaAggregate 实现 DomainEvent 接口，并重写 id 和 occurredOn 方法。

@Service
public class ClickActionApplication extends AbstractApplication {
    @Autowired
    private ClickActionRepository repository;

    @Autowired
    private DomainEventBus domainEventBus;

    public void clickMenu(Long id, String menuId){
        ClickAction clickAction = new ClickAction(id, menuId);
        clickAction.prePersist();
        this.repository.save(clickAction);
        domainEventBus.publish(clickAction);
    }
}

ClickActionApplication 在成功保存 ClickAction 后，使用 DomainEventBus 对事件进行发布。

2.3 订阅领域事件

由什么组件向领域事件注册订阅器呢？大多数请求，由应用服务完成，有时也能够由领域服务进行注册。

因为应用服务是领域模型的直接客户，它是注册领域事件订阅器的理想场所，即在应用服务调用领域方法以前，就完成了对事件的订阅。

基于 ThreadLocal 进行订阅:

public void enable(Long id){
    // 清理以前绑定的 Handler
    DomainEventBusHolder.clean();

    // 注册 EventHandler
    AccountEnableEventHandler enableEventHandler = new AccountEnableEventHandler();
    DomainEventBusHolder.getHandlerRegistry().register(AccountEnabledEvent.class, enableEventHandler);

    Optional<Account> accountOptional = repository.getById(id);
    if (accountOptional.isPresent()) {
        Account account = accountOptional.get();
        // enable 使用 DomainEventBusHolder 直接发布事件
        account.enable();
        repository.save(account);
    }
}

基于实体缓存进行订阅:

@PostConstruct
public void init(){
    // 使用 Spring 生命周期注册事件处理器
    this.domainEventBus.register(AccountEnabledEvent.class, new AccountEnableEventHandler());
}

public void enable(Long id){
    Optional<Account> accountOptional = repository.getById(id);
    if (accountOptional.isPresent()) {
        Account account = accountOptional.get();
        // enable 将事件缓存在 account 中
        account.enable();
        repository.save(account);
        List<DomainEvent> events = account.getEvents();
        if (!CollectionUtils.isEmpty(events)){
            // 成功持久化后，对事件进行发布
            this.domainEventBus.publishAll(events);
        }
    }
}

2.4 处理领域事件

完成事件发布后，让咱们一块儿看下事件处理。

2.4.1 保证聚合间的数据一致性

咱们一般将领域事件用于维护模型的一致性。在聚合建模中有一个原则，就是在一个事务中，只能对一个聚合进行修改，由此产生的变化必须在独立的事务中运行。

在这种状况下，须要谨慎处理的事务的传播性。

应用服务控制着事务。不要在事件通知过程当中修改另外一个聚合实例，由于这样会破坏聚合的一大原则：在一个事务中，只能对一个聚合进行修改。

对于简单场景，咱们可使用特殊的事务隔离策略对聚合的修改进行隔离。具体流程以下：

但，最佳方案是使用异步处理。及每个定义方都在各自独立的事务中修改额外的聚合实例。

事件订阅方不该该在另外一个聚合上执行命令方法，由于这样将破坏“在单个事务中只修改单个聚合实例”的原则。全部聚合实例间的最终一致性必须经过异步方式处理。

详见，异步处理领域事件。

2.4.2 替换批量处理

批处理过程一般须要复杂的查询，而且须要庞大的事务支持。若是在接收到领域事件时，系统就当即处理，业务需求不只获得了更快的知足，并且杜绝了批处理操做。

在系统的非高峰时期，一般使用批处理进行一些系统的维护，好比删除过时数据、建立新的对象、通知用户、更新统计信息等。这些批处理每每须要复杂的查询，并须要庞大的事务支持。

若是咱们监听系统中的领域事件，在接收领域事件时，系统当即处理。这样，本来批量集中处理的过程就被分散成许多小的处理单元，业务须要也能更快的知足，用户能够能够及时的进行下一步操做。

2.4.3 实现事件源模式

对于单个限界上下文中的全部领域事件，为它们维护一个事件存储具备不少的好处。

对事件进行存储能够：

将事件存储做为消息队列来使用，而后将领域事件经过消息设施发布出去。
将事件存储用于基于 Rest 的事件通知。
检查模型命名方法产生结果的历史记录。
使用事件存储来进行业务预测和分析。
使用事件来重建聚合实例。
执行聚合的撤销操做。

事件存储是个比较大的课题，将有专门章节进行讲解。

2.4.4 进行限界上下文集成

基于领域事件的限界上下文集成，主要由消息队列和 REST 事件两种模式。

在此，重心讲解基于消息队列的上下文集成。

在不一样的上下文中采用消息系统时，咱们必须保证最终一致性。在这种状况下，咱们至少须要在两种存储之间保存最终一致性：领域模型所使用的存储和消息队列所使用的持久化存储。咱们必须保证在持久化领域模型时，对于的事件也已经成功发布。若是两种不一样步，模型可能会处于不正确的状态。

通常状况下，有三种方式：

领域模型和消息共享持久化存储。在这种状况下，模型和事件的提交在一个事务中完成，从而保证两种的一致性。
领域模型和消息由全局事务控制。这种状况下，模型和消息所用的持久化存储能够分离，但会下降系统性能。
在领域持久化存储中，建立一个特殊的存储区域用于存储事件（也就是事件存储），从而在本地事务中完成领域和事件的存储。而后，经过后台服务将事件异步发送到消息队列中。

通常状况下，第三种，是比较优雅的解决方案。

在一致性要求不高时，能够经过领域事件订阅器直接向消息队列发送事件。具体流程以下：

对一致性要求高时，须要先将事件存储，而后经过后台线程加载并分发到消息队列。具体流程以下：

2.5 异步处理领域事件

领域事件能够与异步工做流程协同，包括限界上下文间使用消息队列进行异步通讯。固然，在同一个限界上下文中，也能够启动异步处理流程。

做为事件的发布者，不该关心是否执行异步处理。异常处理是由事件执行者决定。

DomainEventExecutor 提供对异步处理的支持。

DomainEventExecutor eventExecutor  =
                new ExecutorBasedDomainEventExecutor("EventHandler", 1, 100);
this.domainEventBus.register(AccountEnabledEvent.class,
        eventExecutor,
        new AccountEnableEventHandler());

异步处理，就意味着放弃数据库事务的 ACID 特性，而选择使用最终一致性。

2.6 内部事件与外部事件

使用领域事件时须要对事件进行区分，以免技术实现的问题。

认识内部事件和外部事件之间的区别相当重要。

内部事件，是一个领域模型内部的事件，不在有界上下文间进行共享。
外部事件，是对外发布的事件，在多个有界上下文中进行共享。

通常状况下，在典型的业务用例中，可能会有不少的内部事件，而只有一两个外部事件。

2.6.1 内部事件

内部事件存在于限界上下文内部，受限界上下文边界保护。

内部事件被限制在单个有界上下文边界内部，因此能够直接引用领域对象。

public interface AggregateEvent<ID, A extends Aggregate<ID>> extends DomainEvent{
    A source();

    default A getSource(){
        return source();
    }
}

好比 AggregateEvent 中的 source 指向发布该事件的聚合。

public class LikeSubmittedEvent extends AbstractAggregateEvent<Long, Like> {
    public LikeSubmittedEvent(Like source) {
        super(source);
    }

    public LikeSubmittedEvent(String id, Like source) {
        super(id, source);
    }
}

LikeSubmittedEvent 类直接引用 Like 聚合。

2.6.2 外部事件

外部事件存在于限界上下文间，被多个上下文共享。

通常状况下，外部事件，只做为数据载体存在。经常采用平面结构，并公开全部属性。

@Data
public class SubmittedEvent {
    private Owner owner;
    private Target target;
}

SubmittedEvent 为扁平化结构，主要是对数据的封装。

因为外部事件被多个上下文共享，版本管理就显得很是重要，以免重大更改对其服务形成影响。

3 实现领域事件模式

领域事件是一种通用模式，它的本质是将领域概念添加到发布-订阅模式。

3.1 封装领域事件的“发布-订阅”模式

发布-订阅是比较成熟的设计模式，具备很高的通用性。所以，建议针对领域需求进行封装。

好比直接使用 geekhalo-ddd 相关模块。

定义领域事件：

@Value
public class LikeCancelledEvent extends AbstractAggregateEvent<Long, Like> {
    public LikeCancelledEvent(Like source) {
        super(source);
    }
}

订阅领域事件：

this.domainEventBus.register(LikeCancelledEvent.class, likeCancelledEvent->{
        CanceledEvent canceledEvent = new CanceledEvent();
        canceledEvent.setOwner(likeCancelledEvent.getSource().getOwner());
        canceledEvent.setTarget(likeCancelledEvent.getSource().getTarget());
        this.redisBasedQueue.pushLikeEvent(canceledEvent);
    });

异步执行领域事件：

DomainEventExecutor eventExecutor  =
                new ExecutorBasedDomainEventExecutor("LikeEventHandler", 1, 100);
this.domainEventBus.register(LikeCancelledEvent.class, 
        eventExecutor, 
        likeCancelledEvent->{
            CanceledEvent canceledEvent = new CanceledEvent();
            canceledEvent.setOwner(likeCancelledEvent.getSource().getOwner());
            canceledEvent.setTarget(likeCancelledEvent.getSource().getTarget());
            this.redisBasedQueue.pushLikeEvent(canceledEvent);
    });

3.2 内存总线处理内部事件，消息队列处理外部事件

内存总线简单高效，同时支持同步、异步两个处理方案，比较适合处理繁杂的内部事件；消息队列虽然复杂，但擅长解决服务间通讯问题，适合处理外部事件。

3.3 使用实体缓存领域事件

理论上，只有在业务成功完成后，才应该对外发布事件。所以，将领域事件缓存在实体中，并在完成业务操做后将其进行发布，是一种较好的解决方案。

相比，使用 ThreadLocal 管理订阅器，并在事件 publish 时进行订阅回调，事件缓存方案有明显的优点。

3.4 使用 IOC 容器的事件发布功能

IOC 容器为咱们提供了不少使用功能，其中也包括发布-订阅功能，如 Spring。

一般状况下，领域模型不该该直接依赖于 Spring 容器。所以，在领域中咱们仍然使用内存总线，为其添加一个订阅者，将内存总线中的事件转发到 Spring 容器中。

class SpringEventDispatcher implements ApplicationEventPublisherAware {

    @Autowired
    private DomainEventBus domainEventBus;

    private ApplicationEventPublisher eventPublisher;

    @Override
    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
        this.eventPublisher = applicationEventPublisher;
    }

    @PostConstruct
    public void addListener(){
        this.domainEventBus.register(event->true, event -> {this.eventPublisher.publishEvent(event);});
    }

}

此时，咱们就能够直接使用 Spring 的 EventListener 机制对领域事件进行处理。

@Component
public class RedisBasedQueueExporter {
    @Autowired
    private RedisBasedQueue redisBasedQueue;

    @EventListener
    public void handle(LikeSubmittedEvent likeSubmittedEvent){
        SubmittedEvent submittedEvent = new SubmittedEvent();
        submittedEvent.setOwner(likeSubmittedEvent.getSource().getOwner());
        submittedEvent.setTarget(likeSubmittedEvent.getSource().getTarget());
        this.redisBasedQueue.pushLikeEvent(submittedEvent);
    }


    @EventListener
    public void handle(LikeCancelledEvent likeCancelledEvent){
        CanceledEvent canceledEvent = new CanceledEvent();
        canceledEvent.setOwner(likeCancelledEvent.getSource().getOwner());
        canceledEvent.setTarget(likeCancelledEvent.getSource().getTarget());
        this.redisBasedQueue.pushLikeEvent(canceledEvent);
    }

}

4 小结

领域事件是发生在问题域中的事实，它是通用语言的一部分。
领域事件优先使用发布-订阅模式，会发布事件而且触发相应的事件处理器。
限界上下文内，优先使用内部事件和内存总线；限界上下文间，优先使用外部事件和消息队列。
领域事件使异步操做变得简单。
领域事件为聚合间提供了最终一致性。
领域事件能够将大的批量操做简化为许多小的业务操做。
领域事件能够完成强大的事件存储。
领域事件能够完成限界上下文间的集成。
领域事件是更复杂架构（CQRS）的一种支持。