EventBus最佳实践：Java手撕实现ESB底层事件委托处理机制的思路

以你之姓@ 2024-03-30 14:43 9阅读 0赞

## 一、为什么要实现EventBus？ ##

首先，当一个组织中的人数众多，我们需要将关注点拆分，以避免陷入逻辑混乱和不必要的代码冲突，而且能带来一些可以让组件复用的好处。  
但是如果将一个复杂的问题解耦成若干个小问题之后，如何让它们有机的协作在一起便成为了一个难题。大多数情况，系统内的各个组件是串联在一起，形成`管道过滤器风格`的架构，即上一个组件的输出是下一个组件的输入。但这种风格下又会导致上游组件并不能快速对下游组件的结果做出反馈，所以通常情况下可以采用AOP切面、函数式回调、委托对象、或者利用传址引用的特性来构建统一上下文的方式来解决。  
所以EventBus的实现目标是：  
1、让代码能够串行执行，在执行过程中还可以派生新的事件。  
2、根据事件传播模型，如果发生异常断言则停止事件传播。  
3、能够对组件串行执行的先后顺序进行编排。  
4、有共性的处理机制可以进行封装，执行过程中视其为透明。  
5、通过隐藏一些实现细节，使每一个关注点看起来都非常易于修改，提高效率。

## 二、EventBus和ESB的关系 ##

ESB支持异构环境中的服务、消息，基于事件的交互，想要实现ESB，首先要在集群的各个单体应用中已经实现了EventBus。当EventBus处理过程中某一个Listener将事件序列化发送给另一台服务器进行处理，而接收方会将Event在自身内部进行事件传播，处理结果的成功与否会影响到事件源服务器的事件传播走向，那么这就实现了简单的ESB。

## 三、平台层设计思路 ##

我们以保存和更新数据为例，保存时需要有校验、生成主键（和编码）、生成审计信息、写入数据库等操作，而更新需要有校验、更新审计信息、更新数据库操作。

因为主键、审计信息、写入数据库是平台提供的能力，而校验则有一部分是平台实现，另一部分是业务上实现，所以我们将应用的层级划分为`平台层`和`领域层`，其实除此之外还有应用层、边缘层、感知层，这个层级划分以具体项目的领域为准。

那么首先，我们在平台层构建一个BaseEntity来作为所有实体的基类，并且带上必要的审计信息：

package org.example.platform.base;
    
    import lombok.Data;
    
    import java.util.Date;
    
    @Data
    public class BaseEntity {
        
        private String uuid;
        private Date createTime;
        private String createUser;
        private Date lastModifyTime;
        private String lastModifyUser;
    }

接下来我们要为事件传播模型构建一个上下文的基类，这里可以根据实际情况来填充：

package org.example.platform.context.entity;
    
    public class BaseContext {
        
    }

然后我们划分出整个应用的上下文：

package org.example.platform.context.entity;
    
    import lombok.Data;
    
    @Data
    public class ApplicationContext extends BaseContext {
        
    }

然后因为审计信息会记录当前用户信息，所以需要有一个会话上下文：

package org.example.platform.context.entity;
    
    import lombok.Data;
    
    @Data
    public class SessionContext extends BaseContext{
        
        private String username;
        private ApplicationContext applicationContext;
    }

为了能更方便的操作和创建上下文，我们做一个上下文的工厂来生产上下文（一般来说会话上下文需要根据你HttpServletRequest对象来拿到Session来确定具体数据，而不是直接new出来，这里只做示意）：

package org.example.platform.context;
    
    import org.example.platform.context.entity.ApplicationContext;
    import org.example.platform.context.entity.SessionContext;
    
    public class ContextFactory {
        
        private static ApplicationContext applicationContext=new ApplicationContext();
    
        public static SessionContext getSessionContext(){
        
            // TODO:这里根据实际情况构建会话上下文
            SessionContext context=new SessionContext();
            context.setUsername("admin");
            // 把应用上下文写入会话上下文中，增强灵活性
            context.setApplicationContext(applicationContext);
            return context;
        }
    }

接下来，我们开始写EventBus的事件基类，这里我们定义一个sender作为事件的发起方对象，而context则表示当前Event对象发起时所处的上下文环境：

package org.example.platform.esb.base;
    
    import lombok.Data;
    import org.example.platform.context.entity.BaseContext;
    
    @Data
    public class BaseEvent {
        
        private Object sender;
        private BaseContext context;
    }

然后我们写一个接口，以一种耦合度非常低的方式来声明监听器应该是什么模样的：

package org.example.platform.esb.itf;
    
    import org.example.platform.esb.base.BaseEvent;
    
    public interface IListener<T extends BaseEvent> {
        
        void doEvent(T event);
    }

最后我们写EventBus的主要实现类ESB，这里的reg方法可以将不同的监听注册上来，并且根据其监听Event类型的不同而避免事件传播时通知到所有的监听（比如新增和修改的Event如果定义的是两个不同的类，则它们之间互相会隔离）。在doEvent的时候，为了能够实现切面、回调效果，此时会在事件处理前后寻找sender（发起方）的before和after开头的方法，如果有就调用。

package org.example.platform.esb;
    
    import org.example.platform.esb.base.BaseEvent;
    import org.example.platform.esb.itf.IListener;
    
    import java.lang.reflect.Method;
    import java.lang.reflect.ParameterizedType;
    import java.util.ArrayList;
    import java.util.LinkedHashMap;
    import java.util.List;
    import java.util.Map;
    
    public class ESB {
        
        private static Map<String, List<IListener>> listeners=new LinkedHashMap<>();
    
        public static void doEvent(BaseEvent e) {
        
            String eventType = e.getClass().getName();
    
            Method beforeMethod = findMethod(e, "before");
            if(beforeMethod!=null){
        
                try {
        
                    beforeMethod.invoke(e.getSender(),e);
                } catch (Exception err) {
        
                    throw new RuntimeException(err);
                }
            }
    
            if(listeners.containsKey(eventType)){
        
                for (IListener iListener : listeners.get(eventType)) {
        
                    iListener.doEvent(e);
                }
            }
    
            Method afterMethod = findMethod(e, "after");
            if(afterMethod!=null){
        
                try {
        
                    afterMethod.invoke(e.getSender(),e);
                } catch (Exception err) {
        
                    throw new RuntimeException(err);
                }
            }
        }
    
        private static Method findMethod(BaseEvent e, String prefix) {
        
            if(e.getSender()!=null){
        
                Object sender = e.getSender();
                for (Method method : sender.getClass().getMethods()) {
        
                    if((prefix + e.getClass().getSimpleName()).equals(method.getName())&&
                            method.getParameterCount()==1 &&
                            method.getParameterTypes()[0].getName().equals(e.getClass().getName())
                    ){
        
                        return method;
                    }
                }
            }
            return null;
        }
    
        public synchronized static void reg(IListener ... listenerArray) {
        
            for (IListener iListener : listenerArray) {
        
                Class type = (Class)((ParameterizedType)iListener
                	.getClass()
                	.getGenericInterfaces()[0]
                ).getActualTypeArguments()[0];
                if(!listeners.containsKey(type.getName())){
        
                    listeners.put(type.getName(),new ArrayList<IListener>());
                }
                if(!listeners.get(type.getName()).contains(iListener)){
        
                    listeners.get(type.getName()).add(iListener);
                }
            }
    
        }
    }

## 四、平台层事件、监听和初始化 ##

首先，我们定义新增的事件，在实例化的时候，会从上下文工厂获取最新的上下文：

package org.example.platform.events;
    
    import lombok.Data;
    import org.example.platform.base.BaseEntity;
    import org.example.platform.context.ContextFactory;
    import org.example.platform.esb.base.BaseEvent;
    
    @Data
    public class CreateEvent extends BaseEvent {
        
        private BaseEntity entity;
    
        public CreateEvent(Object sender, BaseEntity entity) {
        
            setContext(ContextFactory.getSessionContext());
            setSender(sender);
            this.entity=entity;
        }
    }

然后，我们如法炮制，定义修改事件：

package org.example.platform.events;
    
    import lombok.Data;
    import org.example.platform.base.BaseEntity;
    import org.example.platform.context.ContextFactory;
    import org.example.platform.esb.base.BaseEvent;
    
    @Data
    public class ModifyEvent extends BaseEvent {
        
        private BaseEntity entity;
    
        public ModifyEvent(Object sender, BaseEntity entity) {
        
            setContext(ContextFactory.getSessionContext());
            setSender(sender);
            this.entity=entity;
        }
    }

光有事件是不够的，我们还需要分开不同的关注点，将监听器写出来，下面这个是新增时验证的逻辑：

package org.example.platform.listener.create;
    
    import org.example.platform.base.BaseEntity;
    import org.example.platform.esb.itf.IListener;
    import org.example.platform.events.CreateEvent;
    
    public class ValidateByCreate implements IListener<CreateEvent> {
        
        @Override
        public void doEvent(CreateEvent event) {
        
            if (event.getEntity() == null) {
        
                throw new RuntimeException("实体不能为空");
            }
    
            BaseEntity entity = event.getEntity();
            if(entity.getUuid()!=null){
        
                throw new RuntimeException("创建时主键不能有值");
            }
        }
    }

下面是新增时生成主键的逻辑：

package org.example.platform.listener.create;
    
    import org.example.platform.base.BaseEntity;
    import org.example.platform.esb.itf.IListener;
    import org.example.platform.events.CreateEvent;
    
    import java.util.UUID;
    
    public class GenPkByCreate implements IListener<CreateEvent> {
        
        @Override
        public void doEvent(CreateEvent event) {
        
            BaseEntity entity = event.getEntity();
            entity.setUuid(UUID.randomUUID().toString().replaceAll("-",""));
        }
    }

下面是新增时生成审计信息的逻辑：

package org.example.platform.listener.create;
    
    import org.example.platform.base.BaseEntity;
    import org.example.platform.context.entity.SessionContext;
    import org.example.platform.esb.itf.IListener;
    import org.example.platform.events.CreateEvent;
    
    import java.util.Date;
    
    public class GenAuditInfoByCreate implements IListener<CreateEvent> {
        
        @Override
        public void doEvent(CreateEvent event) {
        
            // 获取会话上下文
            SessionContext sessionContext= (SessionContext) event.getContext();
    
            // 设置审计信息
            BaseEntity entity = event.getEntity();
            entity.setCreateTime(new Date());
            entity.setCreateUser(sessionContext.getUsername());
            entity.setLastModifyTime(entity.getCreateTime());
            entity.setLastModifyUser(entity.getCreateUser());
        }
    }

新增时写入数据库的逻辑：

package org.example.platform.listener.create;
    
    import com.google.gson.GsonBuilder;
    import org.example.platform.esb.itf.IListener;
    import org.example.platform.events.CreateEvent;
    
    public class InsertByCreate implements IListener<CreateEvent> {
        
        @Override
        public void doEvent(CreateEvent event) {
        
            // TODO:这里填写实际保存的代码
    
            System.out.println("数据创建成功：");
            System.out.println(new GsonBuilder()
                    .setDateFormat("yyyy-MM-dd HH:mm:ss")
                    .setPrettyPrinting()
                    .create()
                    .toJson(event.getEntity())
            );
        }
    }

这个是修改时验证的逻辑：

package org.example.platform.listener.modify;
    
    import org.example.platform.base.BaseEntity;
    import org.example.platform.esb.itf.IListener;
    import org.example.platform.events.ModifyEvent;
    
    public class ValidateByModify implements IListener<ModifyEvent> {
        
        @Override
        public void doEvent(ModifyEvent event) {
        
            if (event.getEntity() == null) {
        
                throw new RuntimeException("实体不能为空");
            }
    
            BaseEntity entity = event.getEntity();
            if(entity.getUuid()==null){
        
                throw new RuntimeException("修改时主键不能为空");
            }
        }
    }

更新最后修改人和最后修改时间的逻辑：

package org.example.platform.listener.modify;
    
    import org.example.platform.base.BaseEntity;
    import org.example.platform.context.entity.SessionContext;
    import org.example.platform.esb.itf.IListener;
    import org.example.platform.events.ModifyEvent;
    
    import java.util.Date;
    
    public class GenAuditInfoByModify implements IListener<ModifyEvent> {
        
        @Override
        public void doEvent(ModifyEvent event) {
        
            // 获取会话上下文
            SessionContext sessionContext= (SessionContext) event.getContext();
    
            // 设置审计信息
            BaseEntity entity = event.getEntity();
            entity.setLastModifyTime(new Date());
            entity.setLastModifyUser(sessionContext.getUsername());
        }
    }

修改后更新数据库的逻辑：

package org.example.platform.listener.modify;
    
    import com.google.gson.GsonBuilder;
    import org.example.platform.esb.itf.IListener;
    import org.example.platform.events.ModifyEvent;
    
    public class UpdateByModify implements IListener<ModifyEvent> {
        
        @Override
        public void doEvent(ModifyEvent event) {
        
            // TODO:这里填写实际更新的代码
    
            System.out.println("数据修改成功：");
            System.out.println(new GsonBuilder()
                    .setDateFormat("yyyy-MM-dd HH:mm:ss")
                    .setPrettyPrinting()
                    .create()
                    .toJson(event.getEntity())
            );
        }
    }

当写完以上逻辑处理的监听后，就可以写平台层的初始化方法了，这时候就能看出来事件编排的好处和魅力了，如果你希望这些逻辑可以动态调整，也可以从数据库获取、用反射机制实例化，并将初始化过程写在Spring Boot的启动生命周期里，也可以自定义注解加上权重，然后用包扫描的方式自动加载：

package org.example.platform;
    
    import org.example.platform.esb.ESB;
    import org.example.platform.listener.create.GenAuditInfoByCreate;
    import org.example.platform.listener.create.GenPkByCreate;
    import org.example.platform.listener.create.InsertByCreate;
    import org.example.platform.listener.create.ValidateByCreate;
    import org.example.platform.listener.modify.GenAuditInfoByModify;
    import org.example.platform.listener.modify.UpdateByModify;
    import org.example.platform.listener.modify.ValidateByModify;
    
    public class Platform {
        
        public static void init(){
        
            /** 按顺序编排创建过程，一般在技术中台（平台层）实现 */
            // 创建过程
            ESB.reg(
                    new ValidateByCreate(), // 校验
                    new GenPkByCreate(), // 生成主键
                    new GenAuditInfoByCreate(), // 生成审计信息
                    new InsertByCreate() // 写入数据库
            );
            // 修改过程
            ESB.reg(
                    new ValidateByModify(), // 校验
                    new GenAuditInfoByModify(), // 更新审计信息
                    new UpdateByModify()
            );
        }
    }

## 五、领域层编写 ##

首先我们定义一个实体Student，继承自BaseEntity，定义必要的业务字段。

package org.example.domain.student;
    
    import lombok.Data;
    import org.example.platform.base.BaseEntity;
    @Data
    public class Student extends BaseEntity {
        
        private String name;
        private String gender;
        private Integer age;
    }

然后我们写Service服务：

package org.example.domain.student;
    
    import org.example.platform.esb.ESB;
    import org.example.platform.events.CreateEvent;
    import org.example.platform.events.ModifyEvent;
    
    public class StudentService{
        
        public void save(Student student) {
        
            if (student.getUuid() == null) {
        
                ESB.doEvent(new CreateEvent(this,student));
            } else {
        
                ESB.doEvent(new ModifyEvent(this,student));
            }
        }
    
        public void beforeCreateEvent(CreateEvent e){
        
            Student student= (Student) e.getEntity();
            if(student.getName()==null)throw new RuntimeException("姓名不能为空");
            if(student.getGender()==null)throw new RuntimeException("姓名不能为空");
            if(student.getAge()==null)throw new RuntimeException("姓名不能为空");
            if(student.getAge()<6)throw new RuntimeException("未到达入学年龄");
            if(student.getAge()>60)throw new RuntimeException("已超过最大入学年龄");
            System.out.println("新增前事件");
        }
    
        public void afterCreateEvent(CreateEvent e){
        
            System.out.println("新增后事件");
        }
    }

这时候你会发现领域层的实现会非常的简单，而且更易扩展，其中的beforeCreateEvent和afterCreateEvent方法就是扩展的回调函数。如果需要事件有返回值，我们在Context里定义一个输出参数即可，而且下游Listener还可以对返回值进行二次加工（比如带出外联表中的名称字段、缓存处理、等等），这种模式可以让很多玩法成为可能。

## 六、测试 ##

最后我们写一下测试的代码，跑跑看：

package org.example;
    
    import org.example.domain.student.Student;
    import org.example.platform.Platform;
    import org.example.domain.student.StudentService;
    
    public class Test 
    {
        
        public static void main( String[] args )
        {
        
            /** 初始化平台层 */
            Platform.init();
    
            /** 模拟一次新增数据和修改数据的过程，一般在业务中台（领域层）实现 */
            try{
        
                Student stu=new Student();
                stu.setName("杨若瑜");
                stu.setAge(33);
                stu.setGender("男");
    
                StudentService studentService = new StudentService();
                studentService.save(stu);
    
                // 模拟5秒后修改,可以看到lastModifyTime的变化
                Thread.sleep(5000);
    
                stu.setName("瑜美科技");
                studentService.save(stu);
    
            }catch (Exception e){
        
                e.printStackTrace();
            }
        }
    }

执行后，输出结果如下：

新增前事件
    数据创建成功：
    {
        
      "name": "杨若瑜",
      "gender": "男",
      "age": 33,
      "uuid": "1e7ae694b9df4a078d8b208057938141",
      "createTime": "2023-01-05 08:47:54",
      "createUser": "admin",
      "lastModifyTime": "2023-01-05 08:47:54",
      "lastModifyUser": "admin"
    }
    新增后事件
    数据修改成功：
    {
        
      "name": "瑜美科技",
      "gender": "男",
      "age": 33,
      "uuid": "1e7ae694b9df4a078d8b208057938141",
      "createTime": "2023-01-05 08:47:54",
      "createUser": "admin",
      "lastModifyTime": "2023-01-05 08:47:59",
      "lastModifyUser": "admin"
    }

这说明我们的EventBus已经可以正常运作并已实现了目的。  
如果我们将参数的age修改为61，则可以正常命中事件传播链上的beforeCreateEvent方法，输出为：

java.lang.RuntimeException: java.lang.reflect.InvocationTargetException
    	at org.example.platform.esb.ESB.doEvent(ESB.java:24)
    	at org.example.domain.student.StudentService.save(StudentService.java:10)
    	at org.example.Test.main(Test.java:22)
    Caused by: java.lang.reflect.InvocationTargetException
    	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
    	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    	at java.base/java.lang.reflect.Method.invoke(Method.java:568)
    	at org.example.platform.esb.ESB.doEvent(ESB.java:22)
    	... 2 more
    Caused by: java.lang.RuntimeException: 已超过最大入学年龄
    	at org.example.domain.student.StudentService.beforeCreateEvent(StudentService.java:22)
    	... 7 more

可见，通过抛出异常的方式，已成功阻断后续的Listener的执行，达到预期目标。

## 七、总结 ##

尽管本文的篇幅很长，但对于架构设计领域来说，这只是一个非常简单的例子。后续你如果有兴趣还可以基于这种思路拓展WebService、WebSocket、RESTful、RMI的事件传播实现、构建消息队列避免峰值调用，从而真正实现ESB。目前这个例子可以实现单Sender发起Event、通过创建新Event实例并激发，从而树状向后传播，稍加修改，也可以实现多线程环境下多Sender的归并发起单一Event。  
Event传播途中所经之处的组件都可以对Event中包含的内容进行读写，但同样也会带来一些问题。如果需要明确程序员的工作边界并避免调用错误的方法，建议在实际使用过程中Event对象构建成充血模型，并适当对某些Getter和Setter设定private属性。  
例子中StudentService的beforeCreateEvent和afterCreateEvent方法对于某些场景来说可能是一个Bad smell，如果有足够的开发周期进行优化的话，建议用实现接口的方式来实现这两个方法。否则不明真相的程序员在接到二手代码的时候很可能因为静态审查误以为这两个方法没有被引用，从而将其删除。  
综上所述，EventBus是一种非常强有力的“胶水”，可以将不同的业务逻辑片段穿起来，按需进行调用，并可以随时进行扩展。