领域驱动设计 事件驱动与CQRS 详解领域驱动设计之事件驱动与CQRS

package domain.event;

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

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

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

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

/*
* 一个关于比赛的充血模型例子
* 贫血模型会构造一个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;

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

@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;


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