事件驱动模型实践

　　前言
　　Spring的框架中还是在日常MVC代码的编写过程中，巧用事件驱动模型都能很好的提高代码的可维护性。
　　因此，本文将对DDD中使用事件驱动模型建立与踩坑做一个系统性的介绍。从应用层面出发，帮助大家更好的去进行架构迁移。
　　事件驱动模型
　　为什么需要事件驱动模型
　　一个框架，一门技术，使用之前首先要清楚，什么样的业务场景需要使用这个东西。为什么要用跟怎么样把他用好更加重要。
　　假设我们现在有一个比较庞大的单体服务的订单系统，有下面一个业务需求：创建订单后，需要下发优惠券，给用户增长积分
　　先看一下，大多数同学在单体服务内的写法。 //在orderService内部定义一个放下 @Transactional(rollbackFor = Exception.class) public void createOrder(CreateOrderCommand command){   //创建订单   Long orderId = this.doCreate(command);   //发送优惠券   couponService.sendCoupon(command,orderId);   //增长积分   integralService.increase(command.getUserId,orderId); }
　　上面这样的代码在线上运行会不会有问题？不会！
　　那为什么要改呢？
　　原因是，业务需求在不断迭代的过程中，与当前业务非强相关的主流程业务，随时都有可能被替换或者升级。
　　双11大促，用户下单的同时需要给每个用户赠送几个小礼品，那你又要写一个函数了，拼接在主方法的后面。双11结束，这段要代码要被注释。有一年大促，赠送的东西改变，代码又要加回来…
　　来来回回的，订单逻辑变得又臭又长，注释的代码逻辑很多还不好阅读与理解。
　　如果用了事件驱动模型，那么当第一步创建订单成功之后，发布一个创建订单成功的领域事件。优惠券服务，积分服务，赠送礼品等等监听这个事件，对监听到的事件作出相应的处理。
　　事件驱动模型代码 //在orderService内部定义一个放下 @Transactional(rollbackFor = Exception.class) public void createOrder(CreateOrderCommand command){   //创建订单   Long orderId = this.doCreate(command);   publish(orderCreateEvent); }  //各个需要监听的服务 public void handlerEvent(OrderCreateEvent event){ //逻辑处理 }
　　代码解耦，高度符合开闭原则 事件驱动模型选型
　　spring中的事件驱动机制
　　spring在4.2之后提供了@EventListener注解，让我们更便捷的使用监听。
　　了解过spring启动流程的同学都知道，Spring容器刷新的时候会发布ContextRefreshedEvent事件，因此若我们需要监听此事件，直接写个监听类即可。 @Slf4j @Component public class ApplicationRefreshedEventListener implements   ApplicationListener {      @Override     public void onApplicationEvent(ContextRefreshedEvent event) {         //解析这个事件，做你想做的事，嘿嘿     } }
　　同样的我们也可以自己来定义一个事件，通过ApplicationEventPublisher发送。  //领域事件基类  @data  @NoArgsConstructor public abstract class BaseDomainEvent implements Serializable {       //领域事件id      private String demandId;      //发生时间      private LocalDateTime occurredOn;      //领域事件数据      private T data;      public BaseDomainEvent(String demandId, T data) {         this.demandId = demandId;         this.data = data;         this.occurredOn = LocalDateTime.now();     } }
　　定义统一的业务总线发送事件 //领域事件发布接口  public interface DomainEventPublisher {      /*      * 发布事件      *      * @param event 领域事件      */      void publishEvent(BaseDomainEvent event);  }//领域事件发布实现类 @Component @Slf4j public class DomainEventPublisherImpl implements DomainEventPublisher {      @Autowired     private ApplicationEventPublisher applicationEventPublisher;      @Override     public void publishEvent(BaseDomainEvent event) {         log.debug(＂发布事件,event:{}＂, event.toString());         applicationEventPublisher.publishEvent(event);     } }
　　监听事件 @Component @Slf4j public class UserEventHandler {      @EventListener     public void handleEvent(DomainEvent event) {        //doSomething     } }
　　事件驱动之事务管理
　　平时我们在完成某些数据的入库后，发布了一个事件。后续我们进行操作记录在es的记载，但是这时es可能集群响应超时了，操作记录入库失败报错。但是从业务逻辑上来看，操作记录的入库失败，不应该影响到主流程的逻辑执行，需要事务独立。亦或是，如果主流程执行出错了，那么我们需要触发一个事件，发送钉钉消息到群里进行线上业务监控，需要在主方法逻辑中抛出异常再调用此事件。这时，我们如果使用的是@EventListener，上述业务场景的实现就是比较麻烦的逻辑了。
　　为了解决上述问题，Spring为我们提供了两种方式：
　　(1) @TransactionalEventListener注解。
　　(2) 事务同步管理器TransactionSynchronizationManager。
　　本文针对@TransactionalEventListener进行一下解析。
　　我们可以从命名上直接看出，它就是个EventListener，在Spring4.2+，有一种叫做@TransactionEventListener的方式，能够实现在控制事务的同时，完成对对事件的处理。 //被@EventListener标注，表示它能够监听事件 @Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) @Documented@EventListener public @interface TransactionalEventListener {    //表示当前事件跟随消息发送方事务的出发时机，默认为消息发送方事务提交之后才进行处理。   TransactionPhase phase() default TransactionPhase.AFTER_COMMIT;     //true时不论发送方是否存在事务均出发当前事件处理逻辑    boolean fallbackExecution() default false;     //监听的事件具体类型，还是建议指定一下，避免监听到子类的一些情况出现    @AliasFor(annotation = EventListener.class, attribute = ＂classes＂)    Class<?>[] value() default {};     //指向@EventListener对应的值@AliasFor(annotation = EventListener.class, attribute = ＂classes＂)    Class<?>[] classes() default {};     //指向@EventListener对应的值    String condition() default ＂＂;  } public enum TransactionPhase {    // 指定目标方法在事务commit之前执行    BEFORE_COMMIT,     // 指定目标方法在事务commit之后执行     AFTER_COMMIT,      // 指定目标方法在事务rollback之后执行     AFTER_ROLLBACK,     // 指定目标方法在事务完成时执行，这里的完成是指无论事务是成功提交还是事务回滚了    AFTER_COMPLETION   }
　　我们知道，Spring的事件监听机制（发布订阅模型）实际上并不是异步的（默认情况下），而是同步的来将代码进行解耦。而@TransactionEventListener仍是通过这种方式，但是加入了回调的方式来解决，这样就能够在事务进行Commited，Rollback…等时候才去进行Event的处理，来达到事务同步的目的。 实践及踩坑
　　针对是事件驱动模型里面的@TransactionEventListener与@EventListener假设两个业务场景。
　　新增用户，关联角色，增加关联角色赋权操作记录。 统一事务：上述三个操作事务一体，无论哪个发生异常，数据统一回滚。 独立事务：上述三个操作事务独立，事件一旦发布，后续发生任意异常均不影响。
　　统一事务
　　用户新增 @Service @Slf4j public class UserServiceImpl implements UserService {      @Autowired     DomainEventPublisher domainEventPublisher;      @Transactional(rollbackFor = Exception.class)     public void createUser(){         //省略非关键代码save(user);         domainEventPublisher.publishEvent(userEvent);     } }
　　用户角色关联 @Component @Slf4j public class UserEventHandler {      @Autowired     DomainEventPublisher domainEventPublisher;      @Autowired     UserRoleService userRoleService;      @EventListener     public void handleEvent(UserEvent event) {         log.info(＂接受到用户新增事件：＂+event.toString());         //省略部分数据组装与解析逻辑         userRoleService.save(userRole);         domainEventPublisher.publishEvent(userRoleEvent);     }  }
　　用户角色操作记录 @Component @Slf4j public class UserRoleEventHandler {      @Autowired     UserRoleRecordService userRoleRecordService;      @EventListener     public void handleEvent(UserRoleEvent event) {         log.info(＂接受到userRole事件：＂+event.toString());         //省略部分数据组装与解析逻辑         userRoleRecordService.save(record);     }  }
　　以上即为同一事务下的一个逻辑，任意方法内抛出异常，所有数据的插入逻辑都会回滚。
　　给出一下结论，@EventListener标注的方法是被加入在当前事务的执行逻辑里面的，与主方法事务一体。
　　踩坑1
　　严格意义上来说这里不算是把主逻辑从业务中拆分出来了，还是在同步的事务中，当然这个也是有适配场景的，大家为了代码简洁性与函数级逻辑清晰可以这么做。但是这样做其实不是那么DDD，DDD中应用服务的一个方法即为一个用例，里面贯穿了主流程的逻辑，既然是当前系统内强一致性的业务，那就应该在一个应用服务中体现。当然这个是属于业务边界的。举例的场景来看，用户与赋权显然不是强一致性的操作，赋权失败，不应该影响我新增用户，所以这个场景下做DDD改造，不建议使用统一事务。
　　踩坑2 @Component @Slf4j public class UserEventHandler {      @Autowired     DomainEventPublisher domainEventPublisher;      @Autowired     UserRoleService userRoleService;      @EventListener     @Async     public void handleEvent(UserEvent event) {         log.info(＂接受到用户新增事件：＂+event.toString());         //省略部分数据组装与解析逻辑         userRoleService.save(userRole);         domainEventPublisher.publishEvent(userRoleEvent);         throw new RuntimeException(＂制造一下异常＂);     }  }
　　发现，用户新增了，用户角色关联关系新增了，但是操作记录没有新增。第一个结果好理解，第二个结果就奇怪了把，事件监听里面抛了异常，但是居然数据保存成功了。
　　这里其实是因为UserEventHandler的handleEvent方法外层为嵌套@Transactional，userRoleService.save操作结束，事务就提交了，后续的抛异常也不影响。为了保持事务一致，在方法上加一个@Transactional即可。
　　独立事务
　　@EventListener作为驱动加载业务分散代码管理还挺好的。但是在DDD层面，事务数据被杂糅在一起，除了问题一层层找也麻烦，而且数据捆绑较多，还是比较建议使用@TransactionalEventListener
　　用户新增 @Service@Slf4j public class UserServiceImpl implements UserService {      @Autowired     DomainEventPublisher domainEventPublisher;      @Transactional(rollbackFor = Exception.class)     public void createUser(){         //省略非关键代码save(user);         domainEventPublisher.publishEvent(userEvent);     } }
　　用户角色关联 @Component@Slf4j public class UserEventHandler {      @Autowired     DomainEventPublisher domainEventPublisher;      @Autowired     UserRoleService userRoleService;      @TransactionalEventListener     public void handleEvent(UserEvent event) {         log.info(＂接受到用户新增事件：＂+event.toString());         //省略部分数据组装与解析逻辑         userRoleService.save(userRole);         domainEventPublisher.publishEvent(userRoleEvent);     } }
　　用户角色操作记录 @Component @Slf4j public class UserRoleEventHandler {      @Autowired     UserRoleRecordService userRoleRecordService;      @TransactionalEventListener     public void handleEvent(UserRoleEvent event) {         log.info(＂接受到userRole事件：＂+event.toString());         //省略部分数据组装与解析逻辑         userRoleRecordService.save(record);     }  }
　　一样的代码，把注解从@EventListener更换为@TransactionalEventListener。执行之后发现了一个神奇的问题， 用户角色操作记录 数据没有入库！！！ protected void commitTransactionAfterReturning(@Nullable TransactionInfo txInfo) {    if (txInfo != null && txInfo.getTransactionStatus() != null) {       if (logger.isTraceEnabled()) {          logger.trace(＂Completing transaction for [＂ + txInfo.getJoinpointIdentification() + ＂]＂);       }       //断点处       txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());    } }
　　配置文件中添加以下配置 logging:level:org:mybatis: debug
　　在上述代码的地方打上断点，再次执行逻辑。
　　注意看接受到用户新增事件之后的日志，SqlSession [org.apache.ibatis.session.defaults.DefaultSqlSession@9af2818] was not registered for synchronization because synchronization is not active说明当前事件是无事务执行的逻辑。再回过头去看一下@TransactionalEventListener，默认配置是在事务提交后才进行事件执行的，但是这里事务都没有，自然也就不会触发事件了。
　　那怎么解决上面的问题呢？
　　其实这个东西还是比较简单的： 可以对监听此事件的逻辑无脑标注@TransactionalEventListener(fallbackExecution = true),无论事件发送方是否有事务都会触发事件。 在第二个发布事件的上面标注一个@Transactional(propagation = Propagation.REQUIRES_NEW)，切记不可直接标注@Transactional，这样因为userService上事务已经提交，而@Transactional默认事务传播机制为Propagation.REQUIRED,如果当前没有事务，就新建一个事务，如果已经存在一个事务，加入到这个事务中。
　　userService中的事务还存在，只是已经被提交，无法再加入，也就是会导致操作记录仍旧无法被插入。 DDD中的事件驱动应用
　　理清楚spring中事件驱动模型之后，我们所要做的就是开始解耦业务逻辑。
　　通过事件风暴理清楚业务用例，设计完成聚合根，划分好业务领域边界，将原先杂糅在service里面的各个逻辑根据聚合根进行： 对于聚合的每次命令操作，都至少一个领域事 件发布出去，表示操作的执行结果 每一个领域事件都将被保存到事件存储中 从资源库获取聚合时，将根据发生在聚合上的 事件来重建聚合，事件的重放顺序与其产生顺序相同 聚合快照：将聚合的某一事件发生时的状态快 照序列化存储下来。