关于事件驱动架构在工作中的一些想法及实践

2021-12-19 本文已影响0人 javacoo

生活中每天都在发生各种各样的事件，有的被我们关注，有的被我们忽略...

人的精力是有限的，如果对于各种纷繁复杂的事件不加以甄别，筛选，过滤，而一股脑儿的全盘接收，在那些无关紧要的事件上浪费太多的时间与精力，最终只会事倍功半，得不偿失。所以我们只关注我们关心的，重要的事件，当事件发生时才予以相应的回应或动作：我们关注的事件发生，触发了我们的下一步动作------我们的生活不正是由一个一个的事件驱动着前行的吗？：）

应用系统亦是如此，在系统设计时根据不同的需求和使用场景有很多种架构风格可供选择，比如数据流风格、调用/返回风格、独立构件风格、虚拟机风格、仓库风格等：

数据流风格：包括批处理序列风格与管道-过滤器风格，其每一步处理都是独立，顺序执行的，适用于简单的线性流程。
调用/返回风格：包括主程序/子程序、数据抽象、面向对象、层次结构风格，其主要思想是将复杂的大系统分解为小系统，以降低复杂度，增加可修改性。
独立构件风格：包括进程通信和事件驱动系统（隐式调用）风格，其特点是每个构件都是独立的个体，它们之间不直接通信，以降低耦合度，提高灵活性。
虚拟机风格：包括解释器和基于规则的系统风格，其特点是具有良好的灵活性。
仓库风格：包括数据库系统、超文本系统、黑板系统风格，其以数据为中心，善于管理数据信息，适合大量数据、逻辑复杂的系统，除此之外，还有DSSA、REST、分布式等架构风格。

一个业务系统可能是多种架构风格的综合运用，根据进件业务的特点，在设计进件业务模块的时候，可以考虑采用独立构件风格中的事件驱动系统（隐式调用）风格，或者事件驱动架构风格。

接下来结合事件驱动架构谈谈我对进件业务的一些理解和想法...

进件业务模型中的领域事件

领域事件是领域驱动设计中的一个概念，用于捕获我们所建模的领域中所发生过的事情。领域事件本身也作为通用语言的一部分成为包括领域专家在内的所有项目成员的交流用语。比如，在客户申请个人消费贷款过程中，我们可能会说“当客户填写完贷款申请信息并提交成功之后，便进入了贷款申请审批流程”，此时的“客户提交贷款申请成功”便是一个领域事件。

当然，并不是所有发生过的事情都可以成为领域事件。一个领域事件必须对业务有价值，有助于形成完整的业务闭环，也即一个领域事件将导致进一步的业务操作。继续以个人消费贷款为例，假如我们为客户信息系统建模，当客户填写完贷款申请信息并提交成功之后，将产生“客户提交贷款申请成功”的事件，如果你关注的是贷款申请，那么此时的“客户提交贷款申请成功”便是一个典型的领域事件，因为它将用于触发下一步:“为客户建立系统用户体系”的操作；但是如果你建模的是贷后催收系统，那么此时的“客户提交贷款申请成功”事件，便不是我们关心的事情了——你不可能在客户未成功贷款之前，更无逾期的情况下，向客户催收，喊客户还钱对吧，而”客户账单日未还款，且已逾期“才是催收系统应该关注的事件。

事件订阅.png

在DDD中有一条原则：一个业务用例对应一个事务，一个事务对应一个聚合根，也即在一次事务中，只能对一个聚合根进行操作。但是在实际应用中，我们经常发现一个用例需要修改多个聚合根的情况，并且不同的聚合根还处于不同的限界上下文中。比如，当客户贷款申请提交成功后，就会为客户建立相应的用户体系。这里的贷款申请行为可能被建模为一个贷款申请订单对象，而贷款申请信息中的客户相关信息可以建模成客户对象，订单和客户均为聚合根，并且分别属于订单系统和客户系统。显然，我们需要在订单和客户之间维护数据一致性，然而在同一个事务中同时更新两者又违背了DDD设计原则，并且此时需要在两个不同的系统之间采用重量级的分布式事务（Distributed Transactioin，也叫XA事务或者全局事务）。另外，这种方式还在贷款订单系统和客户信息系统之间产生了强耦合。通过引入领域事件，我们可以很好地解决上述问题。总之领域事件给我们带来以下好处：
1. 解耦微服务（限界上下文）
2. 帮助我们深入理解领域模型
3. 提供审计和报告的数据来源
4. 迈向事件溯源（Event Sourcing）和CQRS等
还是以上面的个人消费贷款为例，引入领域事件前，客户填写完贷款申请信息并提交成功之后，进件系统调用相关系统服务进行业务处理，如订单系统，客户系统，账财系统等，这些系统与业务系统都是强耦合的，交互模型如下图：

原进件系统.png

引入领域事件后，客户填写完贷款申请信息并提交成功，进件系统将发出一个“客户提交贷款申请”的领域事件，存储并发布到消息系统中，此时贷款申请便完成了。接下来系统将此领域事件推送到订阅了此事件的各个业务系统系统，进入后续处理流程，比如客户信息系统订阅了消息系统中的“客户提交贷款申请”事件，当事件到达时进行处理，提取事件中的客户信息，再调用自身的服务，建立用户体系。风控系统则根据申请信息，查询客户个人征信，贷款信息等，根据风控模型，决定客户是否满足贷款条件。账财系统则根据申请信息中的贷款信息，进行额度试算，还款计划试算等操作。。。可以看到，此时的进件系统在发送了事件之后，整个用例操作便结束了，根本不用关心是谁收到了事件或者对事件做了什么处理。事件的消费方可以是客户信息系统，也可以是任何一个对事件感兴趣的第三方，比如订单系统，账财系统等。由此，各个微服务之间的耦合关系便解开了。此时各个微服务之间不再是强一致性，而是基于事件的最终一致性。下图是基于领域事件的交互模型：

重构后系统.png

进件系统引入业务领域事件框架(BizEvents-Framework)，其他系统接入BizEvent-SDK，便实现了领域事件的订阅与推送。上图仅仅只是展示领域事件解耦业务微服务的能力，然而现实业务中的情况更为复杂，上图中的账财，风控，客户信息，订单等系统属于基础服务系统，不应该包含太多业务逻辑，他们应该只提供基础服务API供各个业务系统调用，而不必关心具体业务，由各个业务系统整合基础服务API完成最终服务。于是便有了下面的交互模型：

交互模型2.png

现在的交互的逻辑和目前系统现实情况比较贴近，可以在较小的代价下完成现有系统的重构，且可以快速接入新的业务系统。

从上图可以看到进件系统的业务职责进一步抽离到各个业务系统中，只需要关进件事件的订阅及推送，但是此时我们关注的还是进件这一单一领域，我们能不能更进一步，让系统更具通用性，普适性？答案是肯定的：理想状态应该是有一个跨业务领域事件总线（BizEvent-Bus）来统一处理领域事件的发布，订阅与推送，各个系统产生业务领域事件，发布到BizEvent-Bus，同时又从BizEvent-Bus订阅感兴趣的业务领域事件，交互模型如下：

BizEvent-Bus.png

这是理想模型，也是框架后续演进的方向和目标。

接下来谈谈我的一些设计吧...

业务领域事件框架(BizEvents-Framework)

事件驱动架构（Event Driven Architecture，EDA）一个事件驱动框架（EDA）定义了一个设计和实现一个应用系统的方法学，在这个系统里事件可传输于松散耦合的组件和服务之间。一个事件驱动系统典型地由事件消费者和事件产生者组成。事件消费者向事件管理器订阅事件，事件产生者向事件管理器发布事件。当事件管理器从事件产生者那接收到一个事件时，事件管理把这个事件转送给相应的事件消费者。如果这个事件消费者是不可用的，事件管理者将保留这个事件，一段间隔之后再次转送该事件消费者------百度百科

一，简介

业务领域事件框架是一个基于事件驱动架构的基础设施，是一个可以最大程度减少耦合度，很好地扩展与适配不同类型的服务组件。框架分为BizEvents-Framework和BizEvent-SDK两部分，分别扮演服务端和客户端组件的角色，集成BizEvents-Framework能够让应用拥有领域事件的生成，存储，发布，过滤与推送的能力。集成BizEvent-SDK能够让应用拥有订阅目标系统领域事件，并收到目标系统推送领域事件的能力。
业务领域事件框架源于我的另一篇文章《基于RabbitMQ+XXLJob+EventBus的进件平台设计与实现》，当时由于一些原因，进件平台项目未能开发完成就暂停了，自己便做了一个阶段性项目结项，总结了项目的经验教训，于是便有了上面这篇文章。虽然项目暂停了，但是对进件平台的思考从未间断，结合工作中遇到的实际问题，发现的项目痛点与弊端，经过反复的分析，重构，验证，不断完善进件平台架构设计。并在此基础上采取剥离具体业务逻辑，抽象公共处理流程，引入事件机制等，重构进件平台，使其不再仅仅是一个进件平台，更是一个可以最大程度减少耦合度，很好地扩展与适配不同类型的服务组件，是一个基于事件驱动架构的基础设施...慢慢地便有了业务领域事件框架的雏形。

二，交互模型

业务系统发布业务领域事件，分为领域内事件和跨领域事件，领域内事件通过本地事件处理机制处理，跨领域事件通过Runtime首先由Register经过过滤和排序查找订阅了此事件的客户端，再由Stroe对事件进行存储，然后推送到MQ中指定客户端队列，再由MQ客户端通道消费端调用Remote，Remote将事件包装成约定好的协议，通过HTTP/RPC方式推送到对应的客户端，如果推送失败，则重试指定次数，如果任然失败，则后续由Task定时任务处理，客户端则通过集成BizEvent-SDK，实现服务端的接入，接收推送的事件通知。

BizEvent-Framework.png

三，核心特性与能力

基于事件驱动架构

事件驱动架构是一种用于设计应用的软件架构和模型。对于事件驱动系统而言，事件的捕获、通信、处理和持久保留是解决方案的核心结构。事件驱动架构可以最大程度减少耦合度，很好地扩展与适配不同类型的服务组件，因此是现代化分布式应用架构的理想之选。
基于插件化设计

基于xkernel 提供的SPI机制实现，扩展灵活，使用方便。
基于推送的消息传输机制

客户端无需轮询就可以接收更新，事件在到达事件存储后就会通知给客户端，客户端可以随时接收更新，这对于动态数据转换、分析和数据科学处理非常有用。
支持事件过滤

事件的分发与接收支持根据事件类型(eventType)，事件主题(topic)，事件版本(version)以及事件对象中字段值过滤。

四，架构说明

1,BizEvents-Framework

BizEvents-Framework是一个基于事件驱动架构的基础组件，事件驱动架构是一种用于设计应用的软件架构和模型。对于事件驱动系统而言，事件的捕获、通信、处理和持久保留是解决方案的核心结构，业务系统只需依赖BizEvents-Framework基础组件，便可拥有领域事件的生成，存储，发布与推送的能力。

Runtime 整合了以插件化的形式运行的相关插件：

plugin.png

插件基于xkernel 提供的SPI机制实现，扩展非常方便，大致步骤如下：

实现插件扩展接口：

如实现注册插件接口，com.javacoo.events.register.api.service.BizEventRegistryService
```
@Spi(value = BizEventRegistryConfig.DEFAULT_IMPL)
public interface BizEventRegistryService {
    ...
}
```
基于配置文件实现：com.javacoo.events.register.service.BizEventFileRegistryService
```
@Slf4j
public class BizEventFileRegistryService implements BizEventRegistryService {
    ...
}
```
配置自定义插件扩展接口:

plugin-file.png

在项目resource目录新建包->META-INF->ext->internal。
创建以插件扩展接口类全局限定名命名的文件，文件内容：实现名称=实现类的全局限定名，如：

文件名=com.javacoo.events.register.api.service.BizEventRegistryService

内容:
```
file=com.javacoo.events.register.service.BizEventFileRegistryService
```

修改配置文件，添加如下内容:

#业务领域事件注册实现，默认内部实现
biz.event.registry.plugin.impl = file

Stroe插件：主要负责事件的存储，支持对接多种事件存储机制，默认实现：JPA

类结构如下：

store.png

系统第一次启动将创建务领域事件信息表，业务领域事件MQ异常信息表，业务领域事件任务表，插件通过业务领域事件持久化服务对外提供服务。

MQ插件：主要负责事件的发布与订阅，支持对接多种消息中间件，默认实现：RabbitMQ。

类结构如下：

mq.png

基于RabbitMQ 的延迟队列+死信队列实现进件信息的可靠投递与消费，支持根据不同分组生成不同分组交换机，路由KEY和队列，以实现不同分组业务处理的隔离：

分组消费者接口：接口继承ChannelAwareMessageListener ，并新增了启动，停止监听功能，方便后续对不同渠道监听器的灵活控制。
- 分组消费者实现类：实现了onMessage方法，定义了消息处理的主流程：接收消息->转换消息为MessageWrapper对象->获取message对象->调用抽象方法process执行消息推送->如果成功则更新状态为成功->如果失败则调用消息重试方法->如果发送消息过程中出现异常，则调用消息重试方法->最后此条消息要么消费成功，要么从回消息队列。
- 分组消费工厂：应用启动时，根据接入的分组信息动态创建分组消费者对象并启动监听
- 动态消费者容器工厂：实现了FactoryBean接口getObject方法，定义了工作，重试相关交换机，队列，实现了根据接入渠道编号生成对应的交换机和队列，路由key，如：
"分组_work_direct_exchange"，"分组_work_queue","分组_work_routing_key"

并动态申明到RabbitAdmin 中，最后构建SimpleMessageListenerContainer对象并设置相关参数

Remote插件：主要负责向客户端推送订阅的领域事件，支持HTTP和RPC模式。默认实现分别是：Fegin和Dubbo。

类结构如下：

Remote.png

为MQ插件提供远程调用服务，实现了HTTP和RPC（dubbo）方式调用：

业务领域远程调用服务：定义业务领域远程调用服务方法

业务领域远程调用服务抽象实现类：主要实现了业务领域远程调用服务，定义了抽象执行通知方法，由子类具体实现，并基于Sentinel实现了熔断机制。

业务领域远程调用服务工厂：实现了HTTP插件和RPC插件的初始化

HTTP业务领域事件远程调用服务：依赖远程服务组件实现远程服务的调用

Dubbo服务帮助类：此类启动时根据分组配置参数初始化分组->业务领域事件通知服务MAP，主要是根据分组RPC配置参数对dubbo的ReferenceConfig对象进行配置，实现不同分组绑定不同的ReferenceConfig对象，从而达到根据分组调用不同实现的目的。

RPC业务领域事件远程调用服务：依赖Dubbo服务帮助类，根据分组信息找到对应的远程服务并调用方法。

Task插件：主要负责定时从推送失败消息表中获取推送失败的领域事件消息，重新发送到MQ中，继续推送到指定客户端。默认实现：XXLJob

类结构如下：
Task.png

通过定时任务获取推送异常的业务事件信息，重新推送到MQ中，执行推送流程，默认采用XXLJOB：

JOB任务抽象基类：定义JOB执行流程，支持单任务，多任务的串行，并行执行。
业务领域事件任务处理器接口XXLJob实现类：继承JOB任务抽象基类，实现抽象方法，完成任务执行。

Register插件：主要负责客户端系统注册业务领域事件，默认实现：基于配置文件

类结构如下：

Registry.png

通过客户端系统注册信息实现事件的过滤，达到精准推送，按需推送的目的：

业务领域事件注册配置：配置领域事件信息

业务领域事件信息：主要定义了业务领域事件的相关信息

2,BizEvent-SDK

BizEvent-SDK是与BizEvents-Framework配套的客户端SDK，事件驱动的体系结构中，客户端无需轮询就可以接收更新，事件在到达事件存储后就会通知给客户端，客户端可以随时接收更新，这对于动态数据转换、分析和数据科学处理非常有用。其他业务系统只需集成SDK，便可拥有订阅目标系统领域事件，并收到目标系统推送领域事件的能力。类结构图如下：

BizEventDispatcher.png

接下来分两部分来说明：

SDK初始化

系统集成BizEvent-SDK ，系统启动中根据SDK包中spring.factories自动装配相关组件（根据接入配置（见安装教程->客户端），初始化如 HTTP和RPC(dubbo)，过滤器，解析器等），spring.factories配置如下：

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
  com.javacoo.events.client.config.BizEventClientConfig,\
  com.javacoo.events.client.rpc.dubbo.config.DubboConfig,\
  com.javacoo.events.client.rpc.dubbo.DubboAutoConfiguration,\
  com.javacoo.events.client.rpc.dubbo.CustomDubboAutoConfiguration,\
  com.javacoo.events.client.http.HttpAutoConfiguration,\
  com.javacoo.events.client.BizEventListenerRegistrar

BizEventClientConfig：完成了过滤器，解析器的初始化

解析器：默认实现基于Spring SpelExpressionParser实现。EL表达式介绍及使用文档参见 Spring官网。这里需要注意是，SDK约定是基于对象方式访问，对象名称为eventObject，具体使用参见使用说明->客户端，关键代码如下：

/**
  * 获取上下文
  */
 private EvaluationContext getContext(Object root){
        //设置上下文
        EvaluationContext context = new StandardEvaluationContext();
        context.setVariable("eventObject",root);
        return context;
    }

过滤器：过滤器基于自定义函数式接口BizEventFilter实现，定义了匹配方法及辅助方法，支持并操作，或操作，非操作，可实现不同规则下过滤器的灵活组装。默认提供了事件类型过滤器，事件主题过滤器，事件版本过滤器，事件对象参数过滤器，其中前三个过滤器默认实现是简单等值比较，事件对象参数过滤器则是通过解析器解析领域事件中事件对象，获取对象中指定字段的值，与配置的值比较，支持正则表达式匹配。这些过滤器最终通过事件过滤器规则对象组合在一起，默认规则是并操作，关键代码如下：

业务领域事件过滤接口

@FunctionalInterface
public interface BizEventFilter<U,T>{
    /**
     * 匹配
     */
    boolean matcher(U u,T t);
    /**
     * 并操作
     */
    default BizEventFilter<U,T> and(BizEventFilter<? super U,? super T> other) {
        Objects.requireNonNull(other);
        return (u,t) -> matcher(u,t) && other.matcher(u,t);
    }
    /**
     * 或操作
   */
    default BizEventFilter<U,T> or(BizEventFilter<? super U,? super T> other) {
      Objects.requireNonNull(other);
        return (u,t) -> matcher(u,t) || other.matcher(u,t);
    }
    /**
     * 非操作
     */
    default BizEventFilter<U,T> negate() {
        return (u,t) -> !matcher(u,t);
    }
}

事件对象参数过滤器

@Slf4j
public class ParamBizEventFilter implements BizEventFilter<BizEventReq, BizEventListenerInfo> {
    /**
     * 匹配
     */
    @Override
    public boolean matcher(BizEventReq bizEventReq, BizEventListenerInfo bizEventListenerInfo) {
        Objects.requireNonNull(bizEventReq);
        Objects.requireNonNull(bizEventReq.getEventObject());
        Objects.requireNonNull(bizEventListenerInfo);
        log.info("开始表达式匹配:{}", bizEventListenerInfo.getParamEL());
        if(StringUtils.isBlank(bizEventListenerInfo.getParamEL())){
            return true;
        }
        //不支持表达式
        if(!ExpressionParserHolder.getSpelExpressionParser().isPresent()){
            log.warn("不支持表达式匹配");
            return false;
        }
        //获取表达式解析器
        ExpressionParser expressionParser = ExpressionParserHolder.getSpelExpressionParser().get();
        //获取参数class对象
      Class paramClass = bizEventListenerInfo.getTargetMethod().getParameterTypes()[0];
        //获取业务领域事件对象JSON字符串
      final String eventObjectJsonString = JSONObject.toJSONString(bizEventReq.getEventObject());
        //得到目标参数对象
        Object eventObject = JSONObject.parseObject(eventObjectJsonString,paramClass);
        //获取值
        Object valueObj = expressionParser.getValue(bizEventListenerInfo.getEl(),eventObject);
        log.info("取值表达式:{},取值结果:{}", bizEventListenerInfo.getEl(),valueObj);
        if(valueObj == null){
            return false;
        }
        boolean matches = DataUtil.matcher(bizEventListenerInfo.getElValue(),valueObj.toString());
        log.info("值匹配表达式:{},待匹配值:{},匹配结果:{}", bizEventListenerInfo.getElValue(),valueObj.toString(),matches);
        return matches;
    }
}

事件类型过滤器

@Slf4j
public class EventTypeBizEventFilter implements BizEventFilter<BizEventReq, BizEventListenerInfo> {
    /**
     * 匹配
     */
    @Override
    public boolean matcher(BizEventReq bizEventReq, BizEventListenerInfo bizEventListenerInfo) {
        Objects.requireNonNull(bizEventReq);
        Objects.requireNonNull(bizEventListenerInfo);
        log.info("开始事件类型匹配,事件中的类型:{},注册的类型:{}",bizEventReq.getEventType(), bizEventListenerInfo.getEventType());
        if(StringUtils.isBlank(bizEventListenerInfo.getEventType())){
            return false;
        }
        return bizEventReq.getEventType().equals(bizEventListenerInfo.getEventType());
    }
}

事件过滤器规则

@Slf4j
public class SimpleBizEventFilterRule implements BiFunction<BizEventReq, BizEventListenerInfo,Boolean> {
    /** 事件类型过滤器 */
    private BizEventFilter<BizEventReq, BizEventListenerInfo> eventTypeBizEventFilter;
    /** 事件主题过滤器 */
    private BizEventFilter<BizEventReq, BizEventListenerInfo> topicBizEventFilter;
    /** 事件版本过滤器 */
    private BizEventFilter<BizEventReq, BizEventListenerInfo> versionBizEventFilter;
    /** 事件对象参数过滤器 */
    private BizEventFilter<BizEventReq, BizEventListenerInfo> paramBizEventFilter;
    @PostConstruct
    public void init(){
        eventTypeBizEventFilter = new EventTypeBizEventFilter();
        topicBizEventFilter = new TopicBizEventFilter();
        versionBizEventFilter = new VersionBizEventFilter();
        paramBizEventFilter = new ParamBizEventFilter();
        log.info("事件过滤器规则对象初始化完成");
    }
    @Override
    public Boolean apply(BizEventReq bizEventReq, BizEventListenerInfo bizEventListenerInfo) {
        return eventTypeBizEventFilter.and(topicBizEventFilter).and(versionBizEventFilter).and(paramBizEventFilter).matcher(bizEventReq,
            bizEventListenerInfo);
    }
}

DubboConfig，DubboAutoConfiguration，CustomDubboAutoConfiguration三个类，实现了RPC方式接收事件推送通知。

CustomDubboAutoConfiguration自动配置类，在客户端系统未使用dubbo的情况下使用，需要配置dubbo参数，具体配置参见安装教程->客户端代码如下：

@Slf4j
@Configuration(proxyBeanMethods = false)
@DubboComponentScan("com.javacoo.events.client.rpc.dubbo.service")
@EnableConfigurationProperties(value = BizEventDubboConfig.class)
@ConditionalOnClass(BizEventDubboConfig.class)
@ConditionalOnProperty(prefix = BizEventDubboConfig.PREFIX, name=BizEventDubboConfig.CUSTOM_KEY,havingValue = BizEventDubboConfig.CUSTOM_TRUE)
public class CustomDubboAutoConfiguration {
    @Autowired
    private DubboConfig dubboConfig;
    @Bean("biz-event-annotation-provider")
    public ApplicationConfig applicationConfig() {
        return dubboConfig.getApplication();
    }
    @Bean("biz-event-registry")
    public RegistryConfig registryConfig() {
        return dubboConfig.getRegistry();
    }

    @Bean("biz-event-protocol")
    public ProtocolConfig protocolConfig() {
        return dubboConfig.getProtocol();
    }

    @Bean("biz-event-centerConfig")
    public ConfigCenterConfig configCenterConfig(){
        return dubboConfig.getConfigCenter();
    }
    @Bean
    public BizEventDispatcher bizEventDispatcher(){
        return new BizEventDispatcher();
    }
}

业务事件通知实现类

/**
 * 业务事件通知实现类
 * <li>基于dubbo实现</li>
 * <li>version:版本，根据约定，需与接入平台保持一致</li>
 * <li>group:分组，根据约定，需与接入平台保持一致</li>
 */
@Slf4j
@DubboService(interfaceClass = IBizEventNoticeService.class, version = "${biz.event.client.rpc.dubbo.service.version}", retries = -1,group = "${biz.event.client.rpc.dubbo.service.group}")
public class DubboBizEventNoticeService implements IBizEventNoticeService {
    @Autowired
    private BizEventPersistence bizEventPersistence;
    @Autowired
    private BizEventDispatcher bizEventDispatcher;
    /**
     * 通知业务领域事件
     */
    @Override
    public BaseResp notice(BizEventReq bizEventReq) {
        log.info("收到业务领域事件通知:{}",bizEventReq);
        try{
            //持久化业务领域事件
            bizEventPersistence.persist(bizEventReq);
            //分发业务领域事件
            bizEventDispatcher.dispatch(bizEventReq);
      }catch (Exception e){
            log.error("业务领域事件分发失败",e);
          return BaseResp.fail(e.getMessage());
        }
        return BaseResp.ok();
    }
}

DubboAutoConfiguration自动配置类，在客户端已经使用dubbo的情况下使用，主要思想是利用ServiceConfig暴露服务，代码如下：

/**
 * dubbo自动配置
 * <li>依赖已有配置，注册服务</li>
 */
@Slf4j
@Configuration(proxyBeanMethods = false)
@AutoConfigureOrder(5)
@EnableConfigurationProperties(value = BizEventDubboConfig.class)
@ConditionalOnClass(value = {BizEventDubboConfig.class})
@ConditionalOnProperty(prefix = BizEventDubboConfig.PREFIX, name=BizEventDubboConfig.CUSTOM_KEY,havingValue = BizEventDubboConfig.CUSTOM_DEFAULT)
public class DubboAutoConfiguration {
    ...
    @Bean
    public BizEventDispatcher bizEventDispatcher(){
        return new BizEventDispatcher();
    }
    @Bean
    public IBizEventNoticeService bizEventNoticeService(){
        IBizEventNoticeService bizEventNoticeService = new DubboBizEventNoticeService();
        ServiceConfig config = new ServiceConfig();
        config.setApplication(applicationConfig);
        List<RegistryConfig> registryConfigs = new ArrayList<>(1);
        registryConfigs.add(registryConfig);
        config.setRegistries(registryConfigs);
        config.setProtocol(protocolConfig);
        config.setConfigCenter(configCenterConfig);
        config.setGroup(bizEventDubboConfig.getService().getGroup());
      config.setVersion(bizEventDubboConfig.getService().getVersion());
        config.setInterface("com.javacoo.events.api.IBizEventNoticeService");
      config.setRef(bizEventNoticeService);
        config.export();
        return bizEventNoticeService;
    }
}

HttpAutoConfiguration自动配置类，完成Http方式接入，实现很简单，按照约定对外暴露/notice/bizEvent接口方法，代码如下：

@Slf4j
@Configuration
@EnableConfigurationProperties(value = HttpConfig.class)
@ConditionalOnClass(HttpConfig.class)
@ConditionalOnProperty(prefix = HttpConfig.PREFIX, value = HttpConfig.ENABLED, matchIfMissing = true)
public class HttpAutoConfiguration {
    /**
     * http配置
     */
    @Autowired
    private HttpConfig HttpConfig;
    @Bean
    public HttpBizEventNoticeController httpBizEventNoticeController() {
        return new HttpBizEventNoticeController();
    }
    @Bean
  public BizEventDispatcher bizEventDispatcher(){
        return new BizEventDispatcher();
    }
}

业务领域事件通知控制器

@Slf4j
@RestController
@RequestMapping("/notice")
public class HttpBizEventNoticeController {
....
    /**
     * 业务领域事件通知
     */
    @RequestMapping("/bizEvent")
  @ResponseBody
    public BaseResp notice(@RequestBody BizEventReq bizEventReq) {
      log.info("HttpBizEventNoticeController收到业务领域事件通知:{}",bizEventReq);
        try{
          //持久化业务领域事件
            bizEventPersistence.persist(bizEventReq);
            //分发业务领域事件
            bizEventDispatcher.dispatch(bizEventReq);
        }catch (Exception e){
            log.error("业务领域事件分发失败",e);
            return BaseResp.fail(e.getMessage());
        }
        return BaseResp.ok();
    }
}

BizEventListenerRegistrar业务领域事件监听对象注册服务：系统集成BizEvent-SDK （配置见安装教程->客户端）,Spring容器启动之后，便会运行业务领域事件监听对象注册服务（BizEventListenerRegistrar），扫描Spring容器中包含@Service注解的服务，找到其中添加了@BizEventListener注解的方法，解析并组装业务领域事件监听信息（包括事件类型，事件主题，事件版本，事件对象参数匹配:EL表达式，目标方法），完成业务事件监听元数据集合的组装，供后续使用。关键代码如下：

业务领域事件监听对象注册服务

@Slf4j
@Component
public class BizEventListenerRegistrar implements ApplicationContextAware {
    /** 业务事件监听元数据集合 */
    private static List<BizEventListenerMetaData> bizEventListenerMetaDataList = new ArrayList<>();
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        //设置上下文
        ApplicationContextProvider.setApplicationContext(applicationContext);
        //扫描Service注解，初始化业务事件监听元数据集合
        initBizEventMetaDataList(Service.class);
    }
    /**
     * 扫描指定注解服务，初始化业务事件监听元数据集合s对象
     * @return: void
     */
    private void initBizEventMetaDataList(Class<? extends Annotation> annClass){
        //查找Service
        Map<String, Object> serviceMap = ApplicationContextProvider.getApplicationContext().getBeansWithAnnotation(annClass);
        for (Map.Entry<String, Object> entry : serviceMap.entrySet()) {
            Class entryClass = AopUtils.getTargetClass(entry.getValue());
            //获取业务领域事件监听信息集合
            List<BizEventListenerInfo> bizEventListenerInfos = Arrays.stream(entryClass.getDeclaredMethods())
                //获取本类 public方法
                .filter(method -> Modifier.isPublic(method.getModifiers()))
                //找到注解所在方法
                .filter(method -> method.isAnnotationPresent(BizEventListener.class))
                //只支持监听一个参数
                .filter(method -> method.getParameterTypes().length == 1)
                //排序
                .sorted(Comparator.comparing(method -> method.getAnnotation(BizEventListener.class).order()))
                //组装
                .map(method -> getBizEventInfo(method))
                .collect(Collectors.toList());
            if(bizEventListenerInfos.isEmpty()){
                continue;
            }
      bizEventListenerMetaDataList.add(BizEventListenerMetaData.builder().beanName(entry.getKey()).bizEventListenerInfos(bizEventListenerInfos).targetClass(entryClass).build());
        }
        log.info("业务领域事件监听对象注册数量:{},对象:{}", bizEventListenerMetaDataList.size(), bizEventListenerMetaDataList);
    }
    /**
     * 获取方法上的注解信息，组装业务领域事件监听信息
     */
    private BizEventListenerInfo getBizEventInfo(Method method){
        BizEventListener bizEventListener = method.getAnnotation(BizEventListener.class);
        return BizEventListenerInfo.builder()
            .eventType(bizEventListener.eventType())
            .topic(bizEventListener.topic())
            .version(bizEventListener.version())
            .paramEL(bizEventListener.paramEl())
          .targetMethod(method)
            .build();
  }
    /**
     * 获取业务事件监听元数据集合
     */
    public static Optional<List<BizEventListenerMetaData>> getBizEventListenerMetaDataList(){
        return bizEventListenerMetaDataList.isEmpty() ? Optional.empty() : Optional.of(bizEventListenerMetaDataList);
    }
}

业务领域事件监听注解

@Inherited
@Documented
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface BizEventListener {
    /**
     * 业务事件类型
     */
    String eventType() default "";
    /**
     * 业务事件主题
   */
    String topic() default "";
  /**
     * 业务事件版本
   */
    String version() default "";
  /**
     * 参数EL匹配表达式->格式:EL表达式=取值
     */
    String paramEl() default "";
    /**
     * 排序
     */
    int order() default 0;
}

至此SDK初始化便完成。

事件处理

通知事件的处理，则通过业务领域事件持久化服务BizEventPersistence和业务领域事件分发器BizEventDispatcher完成，业务领域事件持久化服务完成事件的持久化处理，SDK提供默认持久化服务（什么都不做），客户端可根据业务需求，扩展实现。业务领域事件分发器主要是根据接收到业务领域事件通知信息，从系统启动时收集到的业务事件监听元数据集合中依次调用事件过滤器规则服务SimpleBizEventFilterRule找到匹配的事件监听方法，利用反射机制执行。执行分异步和同步执行，由下发的通知事件中async参数决定，默认是异步的，执行时，具有相同事件匹配规则的方法是按照order自然排序后，顺序执行的，且传递的事件对象是同一个对象，支持修改。代码如下：

@Slf4j
public class BizEventDispatcher {
    /**
     * 事件过滤器规则
     */
    @Autowired
    private SimpleBizEventFilterRule simpleBizEventFilterRule;
    /**
     * 任务执行器
     */
    private TaskExecutor taskExecutor;
    @PostConstruct
    public void init(){
        taskExecutor = new TaskExecutor() {
            ExecutorService executorService = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors(),
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(), Executors.defaultThreadFactory());
            @Override
            public void execute(Runnable task) {
                executorService.execute(task);
            }
        };
    }
    public void dispatch(final BizEventReq bizEventReq){
        //异步执行
        if (bizEventReq.async()) {
            log.info("异步分发业务领域事件");
            taskExecutor.execute(()->doDispatch(bizEventReq));
        } else {
            log.info("同步分发业务领域事件");
            doDispatch(bizEventReq);
        }
    }
    private void doDispatch(final BizEventReq bizEventReq){
       if(!BizEventListenerRegistrar.getBizEventListenerMetaDataList().isPresent()){
            log.warn("未注册事件监听方法:{}",bizEventReq.getEventType());
            return;
        }
        //获取业务事件监听元数据集合
        List<BizEventListenerMetaData> bizEventMetaDataList = BizEventListenerRegistrar.getBizEventListenerMetaDataList().get();
        //获取业务领域事件对象JSON字符串
        final String eventObjectJsonString = JSONObject.toJSONString(bizEventReq.getEventObject());
        //业务领域事件对象Map:相同参数监听方法间传递同一业务领域事件对象
        Map<String,Object> eventObjectMap = new HashMap<>();
        //执行分发
        bizEventMetaDataList.forEach(bizEventListenerMetaData -> {
            //根据spring bean名称查找bean对象
            final Object beanObject = ApplicationContextProvider.getBean(bizEventListenerMetaData.getBeanName());
            bizEventListenerMetaData.getBizEventListenerInfos().parallelStream()
                    .filter(bizEventListenerInfo -> simpleBizEventFilterRule.apply(bizEventReq,bizEventListenerInfo))
                    .forEachOrdered(bizEventListenerInfo -> execute(eventObjectJsonString,eventObjectMap,beanObject,bizEventListenerInfo.getTargetMethod()));
        });
    }
    private void execute(String eventObjectJsonString, Map<String, Object> eventObjectMap, Object beanObject, Method method) {
        Class paramClass = method.getParameterTypes()[0];
        Object eventObject = null;
        if (eventObjectMap.containsKey(paramClass.getName())) {
            eventObject = eventObjectMap.get(paramClass.getName());
        } else {
            eventObject = JSONObject.parseObject(eventObjectJsonString,paramClass);
            eventObjectMap.put(paramClass.getName(), eventObject);
        }
        try {
            method.invoke(beanObject, eventObject);
        } catch (IllegalAccessException e) {
            e.printStackTrace();
            log.error("方法访问异常",e);
        } catch (InvocationTargetException e) {
            e.printStackTrace();
            log.error("方法调用异常",e);
        }
    }
}

3,Protocol

Protocol是BizEvents-Framework和BizEvent-SDK连接的桥梁。

字段	类型	描述
transactionSn	String	交易流水号
eventGroup	String	事件组
eventId	String	事件ID
eventVersion	String	事件版本
eventTopic	String	事件主题
eventType	String	事件类型: 约定：事件类型=事件对象类型
eventObject	Object	事件对象
async	Boolean	是否异步
timestamp	String	事件时间戳

五，安装教程

1，服务端

引入依赖包:

       <dependency>
            <groupId>com.javacoo</groupId>
            <artifactId>bizevents-runtime</artifactId>
            <version>1.0.0</version>
        </dependency>

配置参数：以下是接入的最小配置示例，其中一些基础依赖配置如：RabbitMQ，数据源等未列出，按照与SpringBoot 集成方式配置即可。

## =============================业务领域事件组件配置================开始
## 业务领域事件注册插件配置-默认配置文件方式
# TEST001 TEST001->表示接入的客户端系统ID，event-type->表示注册的业务领域事件
biz.event.registry.plugin.file.groups.TEST001[0].event-type=com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent
# TEST001 TEST001->表示接入的客户端系统ID,topic->事件对象参数EL表达式,格式:EL表达式=取值->取值精准匹配
biz.event.registry.plugin.file.groups.TEST001[0].param-eL=#eventObject.userType=A
# TEST001 TEST001->表示接入的客户端系统ID，event-type->表示注册的业务领域事件
biz.event.registry.plugin.file.groups.TEST001[1].event-type=com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent
# TEST001 TEST001->表示接入的客户端系统ID,topic->事件对象参数EL表达式,格式:EL表达式=取值->取值正则表达式匹配
biz.event.registry.plugin.file.groups.TEST001[1].param-eL=#eventObject.userName=.*徐.*
# TEST001 TEST001->表示接入的客户端系统ID，event-type->表示注册的业务领域事件
biz.event.registry.plugin.file.groups.TEST001[2].event-type=com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent
# TEST001 TEST001->表示接入的客户端系统ID,version->事件版本
biz.event.registry.plugin.file.groups.TEST001[2].version=1.0
# TEST001 TEST001->表示接入的客户端系统ID，event-type->表示注册的业务领域事件
biz.event.registry.plugin.file.groups.TEST001[3].event-type=com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent
# TEST001 TEST001->表示接入的客户端系统ID,topic->事件主题
biz.event.registry.plugin.file.groups.TEST001[3].topic=天下第二
# TEST001 TEST001->表示接入的客户端系统ID，event-type->表示注册的业务领域事件
biz.event.registry.plugin.file.groups.TEST001[4].event-type=com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent
# TEST001 TEST001->表示接入的客户端系统ID,version->事件版本
biz.event.registry.plugin.file.groups.TEST001[4].version=1.0
# TEST001 TEST001->表示接入的客户端系统ID,topic->事件主题
biz.event.registry.plugin.file.groups.TEST001[4].topic=桃花剑神
biz.event.registry.plugin.file.groups.TEST001[5].event-type=com.javacoo.xservice.example.bean.event.ServerExampleEvent
# TEST002 业务领域事件注册配置-配置文件方式-接入：其中：TEST002-表示接入的客户端系统ID，event-type->表示注册的业务领域事件
#biz.event.registry.plugin.file.groups.TEST002[0].event-type=com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent
#biz.event.registry.plugin.file.groups.TEST002[1].event-type=com.javacoo.xservice.example.bean.event.ServerExampleEvent


## 业务领域事件-MQ插件配置-默认RabbitMQ
# TEST001 推送类型配置-RPC
biz.event.mq.plugin.groups.TEST001.remote-type=RPC
# TEST002 推送类型配置-HTTP
biz.event.mq.plugin.groups.TEST002.remote-type=HTTP

## 业务领域事件-远程调用插件配置
#降级配置
biz.event.remote.plugin.grade=2
biz.event.remote.plugin.count=10
# HTTP配置
# TEST001-推送地址
biz.event.remote.plugin.http.groups.TEST001.push-url=http://127.0.0.1:8188
# TEST001-请求超时时间
biz.event.remote.plugin.http.groups.TEST001.socket-timeout=60000

# TEST002-推送地址
biz.event.remote.plugin.http.groups.TEST002.push-url=http://127.0.0.1:8080
# TEST002-请求超时时间
biz.event.remote.plugin.http.groups.TEST002.socket-timeout=60000

# TEST001 RPC配置-zk地址
biz.event.remote.plugin.rpc.groups.TEST001.address=zookeeper://zookeeper01-dev.javacoo.com:22181?backup=zookeeper02-dev.javacoo.com:22181,zookeeper03-dev.javacoo.com:22181
# TEST001 RPC配置-服务分组:格式->{客户端系统ID}_NOTICE_SERVICE
biz.event.remote.plugin.rpc.groups.TEST001.group=TEST001_NOTICE_SERVICE
# TEST001 RPC配置-服务版本
biz.event.remote.plugin.rpc.groups.TEST001.version=1.0.0
# TEST001 RPC配置-请求超时时间
biz.event.remote.plugin.rpc.groups.TEST001.socket-timeout=60000

# TEST002 RPC配置-zk地址
biz.event.remote.plugin.rpc.groups.TEST002.address=zookeeper://zookeeper01-dev.javacoo.com:22181?backup=zookeeper02-dev.javacoo.com:22181,zookeeper03-dev.javacoo.com:22181
# TEST002 RPC配置-服务分组:格式->{客户端系统ID}_NOTICE_SERVICE
biz.event.remote.plugin.rpc.groups.TEST002.group=TEST002_NOTICE_SERVICE
# TEST003 RPC配置-服务版本
biz.event.remote.plugin.rpc.groups.TEST002.version=1.0.0
# TEST004 RPC配置-请求超时时间
biz.event.remote.plugin.rpc.groups.TEST002.socket-timeout=60000

## 业务领域事件-存储插件配置
biz.event.store.plugin.jpa.enabled=true
## JPA配置
spring.jpa.show-sql=true
spring.jpa.generate-ddl=true
spring.jpa.hibernate.ddl-auto=update
spring.jpa.database-platform=org.hibernate.dialect.MySQL5InnoDBDialect

## 业务领域事件-定时任务插件配置
biz.event.task.plugin.impl=xxlJob

## 业务领域事件-定时任务xxljob插件配置
### 调度中心部署跟地址 [选填]：如调度中心集群部署存在多个地址则用逗号分隔。执行器将会使用该地址进行"执行器心跳注册"和"任务结果回调"；为空则关闭自动注册；
biz.event.task.plugin.xxljob.admin-addresses=https://xxl-job-admin-javacoo.com/
### 执行器AppName [选填]：执行器心跳注册分组依据；为空则关闭自动注册
biz.event.task.plugin.xxljob.app-name=bizevent-job
### 执行器IP [选填]：默认为空表示自动获取IP，多网卡时可手动设置指定IP，该IP不会绑定Host仅作为通讯实用；地址信息用于 "执行器注册" 和 "调度中心请求并触发任务"；
biz.event.task.plugin.xxljob.ip=
### 执行器端口号 [选填]：小于等于0则自动获取；默认端口为9999，单机部署多个执行器时，注意要配置不同执行器端口；
biz.event.task.plugin.xxljob.port=9999
### 执行器通讯TOKEN [选填]：非空时启用；
biz.event.task.plugin.xxljob.access-token=
### 执行器运行日志文件存储磁盘路径 [选填] ：需要对该路径拥有读写权限；为空则使用默认路径；
biz.event.task.plugin.xxljob.log-path=/volume_logs
### 执行器日志保存天数 [选填] ：值大于3时生效，启用执行器Log文件定期清理功能，否则不生效；
biz.event.task.plugin.xxljob.log-retention-days=-1
## =============================业务领域事件组件配置================结束

2，客户端

所谓客户端都是相对的，只要集成了BizEvent-SDK我们都叫他客户端，目前SDK支持HTTP方式和RPC(dubbo)方式接入，具体配置如下:

引入依赖包:

       <dependency>
            <groupId>com.javacoo</groupId>
            <artifactId>bizevents-sdk-java</artifactId>
            <version>1.0.0</version>
        </dependency>

配置参数：

http接入：如果是http接入，则无需任何配置，只需要将客户端服务地址和端口告知服务端(目前是通过人工配置方式)，在服务端配置即可，如：http://127.0.0.1:8080

RPC接入：如果是RPC接入，则分为2种情况：

当前客户端系统已经集成dubbo，则参数配置如下:

#=====================业务领域事件SDK rpc-dubbo 接入配置===========
#RPC服务配置
#是否自定义dubbo配置:false->表示系统已经集成dubbo,无需配置dubbo参数
biz.event.client.rpc.dubbo.custom=false
#dubbo服务分组:格式->{系统ID}_NOTICE_SERVICE,系统ID由服务端统一分配
biz.event.client.rpc.dubbo.service.group=TEST002_NOTICE_SERVICE
#dubbo服务版本:当前服务的版本
biz.event.client.rpc.dubbo.service.version=1.0.0

当前客户端系统未集成dubbo，则参数配置如下:

#=====================业务领域事件SDK rpc-dubbo 接入配置===========
#RPC服务配置
#是否自定义dubbo配置:false->表示系统已经集成dubbo,无需配置dubbo参数
biz.event.client.rpc.dubbo.custom=true
#dubbo服务分组:格式->{系统ID}_NOTICE_SERVICE,系统ID由服务端统一分配
biz.event.client.rpc.dubbo.service.group=TEST001_NOTICE_SERVICE
#dubbo服务版本:当前服务的版本
biz.event.client.rpc.dubbo.service.version=1.0.0

#dubbo配置-application
dubbo.application.name = fund-api
dubbo.application.id = fund-api
dubbo.application.qos-enable = false
#dubbo配置-registry
dubbo.registry.id = registry
dubbo.registry.address = zookeeper://zookeeper01-dev.javacoo.com:22181?backup=zookeeper02-dev.javacoo.com:22181,zookeeper03-dev.javacoo.com:22181
#dubbo配置-protocol
dubbo.protocol.name = dubbo
dubbo.protocol.id = dubbo
dubbo.protocol.port = 20887
dubbo.protocol.host = 0.0.0.0
dubbo.protocol.heartbeat = 30
dubbo.protocol.accesslog = /volume_logs/rpc-access.log
#dubbo配置-config-center
dubbo.config-center.timeout = 60000

六，使用说明

1，服务端

服务端使用BizEventPublisher类直接发布业务领域事件即可，如：

服务端发布领域事件类型：com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent，协议如下：

字段	类型	说明
userName	String	用户名
userType	String	用户分类
amount	BigDecimal	金额

对应的领域事件对象：com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class ServerLoanApplyEvent {
    /**
     * 用户名
     */
    private String userName;
    /**
     * 用户分类
     */
    private String userType;
    /**
     * 金额
     */
    private BigDecimal amount;
}

使用BizEventPublisher发布事件，此时可根据具体业务，设置事件的版本，主题

        //发布普通领域事件
        BizEventPublisher.publish(ServerLoanApplyEvent.builder()
            .userName("徐凤年")
            .userType("A")
            .amount(new BigDecimal("30000"))
            .build());
        //发布普通领域事件
        BizEventPublisher.publish(ServerLoanApplyEvent.builder()
            .userName("徐渭熊")
            .userType("B")
            .amount(new BigDecimal("10000"))
            .build());
        //发布普通领域事件
        BizEventPublisher.publish(ServerLoanApplyEvent.builder()
            .userName("温华")
            .userType("C")
            .amount(new BigDecimal("20000"))
            .build());
        //发布带主题的领域事件
        BizEventPublisher.publish(ServerLoanApplyEvent.builder()
            .userName("王仙芝")
            .userType("A")
            .amount(new BigDecimal("100000"))
            .build(),"天下第二");
        //发布带版本的领域事件
        BizEventPublisher.publish(ServerLoanApplyEvent.builder()
            .userName("李淳罡")
            .userType("B")
            .amount(new BigDecimal("200000"))
            .build(),"","1.0");
        //发布带主题+版本的领域事件
        BizEventPublisher.publish(ServerLoanApplyEvent.builder()
            .userName("邓太阿")
            .userType("C")
            .amount(new BigDecimal("300000"))
            .build(),"桃花剑神","1.0");

如发布带主题+版本的领域事件，生成协议JSON格式如下：

{
    "async":false,
    "eventId":"EventId_2021120900000077",
    "eventVersion":"1.0",
    "eventObject":{
        "amount":300000,
        "userType":"C",
        "userName":"邓太阿"
    },
    "eventTopic":"桃花剑神",
    "eventType":"com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent",
    "eventGroup":"TEST001",
    "transactionSn":"TransSn_2021120900000073",
    "timestamp":"20211209150610183"
}

2，客户端

客户端只需在事件处理方法上加上 @BizEventListener注解，填写服务端发布的领域业务事件类型，主题和版本，或者参数过滤条件，即可收到服务端推送的相应业务领域事件。如：

客户端订阅服务端发布的领域事件类型：com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent，则客户端首先按照事件类型协议，定义监听对象参数对象,参数对象字段与协议保持一致：

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class ClientLoanApplyEvent {
    /**
     * 用户名
     */
    private String userName;
    /**
     * 用户分类
     */
    private String userType;

    /**
     * 金额
     */
    private BigDecimal amount;
}

接着定义事件监听方法，如下：

在监听方法上添加@BizEventListener注解，填写服务端发布的领域业务事件类型，主题和版本，或者参数过滤条件

@Slf4j
@Service
public class BizEventTestService {
    /**
     * 指定事件类型+事件版本
     */
    @BizEventListener(eventType = "com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent",version = "1.0",order = 1)
    public void test(ClientLoanApplyEvent clientLoanApplyEvent){
        log.info("BizEventTestService收到版本事件推送:{}", FastJsonUtil.toJSONString(clientLoanApplyEvent));
        clientLoanApplyEvent.setAmount(new BigDecimal("50000"));
    }
    /**
     * 指定事件类型+事件主题
     */
    @BizEventListener(eventType = "com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent",topic = "天下第二",order = 1)
    public void test1(ClientLoanApplyEvent clientLoanApplyEvent){
        log.info("BizEventTestService收到主题事件推送:{}", FastJsonUtil.toJSONString(clientLoanApplyEvent));
        clientLoanApplyEvent.setAmount(new BigDecimal("70000"));
    }
    /**
     * 指定事件类型+事件主题+事件主题
     */
    @BizEventListener(eventType = "com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent",version = "1.0",topic = "桃花剑神",order = 1)
    public void test2(ClientLoanApplyEvent clientLoanApplyEvent){
        log.info("BizEventTestService收到版本+主题事件推送:{}", FastJsonUtil.toJSONString(clientLoanApplyEvent));
    }
    /**
     * 指定事件类型
     */
    @BizEventListener(eventType = "com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent")
    public void test4(ClientLoanApplyEvent clientLoanApplyEvent){
        log.info("BizEventTestService收到事件推送:{}", FastJsonUtil.toJSONString(clientLoanApplyEvent));
    }
    /**
     * 指定事件类型+值正则匹配
     */
    @BizEventListener(eventType = "com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent",paramEl = "#eventObject.userName=.*徐.*")
    public void test5(ClientLoanApplyEvent clientLoanApplyEvent){
        log.info("BizEventTestService收到值正则匹配事件推送:{}", FastJsonUtil.toJSONString(clientLoanApplyEvent));
        try {
            Thread.sleep(2*1000);
        } catch (InterruptedException interruptedException) {
            interruptedException.printStackTrace();
        }
    }
    /**
     * 指定事件类型+值精准匹配
     */
    @BizEventListener(eventType = "com.javacoo.xservice.example.bean.event.ServerLoanApplyEvent",paramEl = "#eventObject.userType=A")
    public void test6(ClientLoanApplyEvent clientLoanApplyEvent){
        log.info("BizEventTestService收到值精准匹配事件推送:{}", FastJsonUtil.toJSONString(clientLoanApplyEvent));
    }
    /**
     * 指定事件类型
     */
    @BizEventListener(eventType = "com.javacoo.xservice.example.bean.event.ServerExampleEvent")
    public void handleClientExampleEvent(ClientExampleEvent clientExampleEvent){
        log.info("BizEventTestService->handleClientExampleEvent收到事件推送:{}", FastJsonUtil.toJSONString(clientExampleEvent));
    }
}

注意事项：

业务领域事件监听方法参数对象字段，需要与服务端推送的业务领域对象字段保持一致，即必须根据服务端发布的业务领域事件类型来定义客户端业务领域事件监听方法参数对象。
SDK约定业务领域事件监听方法所在服务类必须添加@Service注解。

效果

值正则匹配

处理值正则匹配.png

值精准匹配

处理值精准匹配.png

事件类型+事件版本

类型和版本匹配.png

事件类型+事件主题

类型主题匹配.png

事件类型+事件版本+事件主题

类型版本主题匹配.png

事件类型匹配

类型匹配.png

七，后续规划

增加BizEvent-SDK发布事件的能力
增加预警能力
升级框架为BizEvents-Bus

八，问题及局限性

事件的发布，存储，推送MQ，MQ消费，推送都在一个工程，降低了系统的可靠性及可用性：后续可以考虑拆分，分别单独部署。
事件消费，依赖推送是否成功，依赖客户端的处理能力，虽然目前SDK提供了持久化接口，由客户端决定是否持久化，且事件的分发默认是异步的，可以最大程度提高响应速度，但是客户端性能是不可控的，当遇到客户端响应超时，将导致MQ消息堆积，将影响服务端的性能。
目前BizEvents-Framework还是一个雏形，是基于对现有业务的梳理，理解和抽象，结合现有相关系统特点加上自己的一些想法而得出的设计原型。这也算是我对于这段时间工作的梳理和总结，对接下来的工作任务也有参考和指导意义。由于本人涉及到的业务领域有限，其设计存在局限性和不合理的地方，仅供学习和交流，不足之处欢迎吐槽和指正，本人将不胜感激：）

关于事件驱动架构在工作中的一些想法及实践

进件业务模型中的领域事件

业务领域事件框架(BizEvents-Framework)

一，简介

二，交互模型

三，核心特性与能力

四，架构说明

1,BizEvents-Framework

2,BizEvent-SDK

3,Protocol

五，安装教程

1，服务端

2，客户端

六，使用说明

1，服务端

2，客户端

七，后续规划

八，问题及局限性

猜你喜欢

热点阅读