基于Spring事件驱动模式实现业务解耦

基于,spring,事件驱动,模式,实现,业务 · 浏览次数 : 44

小编点评

# 内容生成过程 1. **定义接口**: - BizEventListener:用于注册业务事件 - EventEngine:用于注册事件驱动引擎 - EventEngineTopic:用于定义事件主题 2. **创建实例**: - EventEngineImpl:用于创建事件引擎 3. **配置事件引擎**: - 设置bizSubscribers、async、bizListenerExecutor 4. **注册事件**: - 注册业务事件监听器 - 注册事件驱动引擎监听器 - 注册消息推送引擎监听器 5. **发布事件**: - 发布事件到事件主题 6. **测试**: - 使用@Resource注解注入EventEngine实例 - 使用@GetMapping注解定义注册事件 - 使用@RequestParam注解定义事件参数 7. **排版输出**: - 使用简单的排版格式输出注册事件、事件驱动引擎监听器和消息推送引擎监听器的状态 8. **项目代码结构**: - 项目代码结构调用接口,其中包含bizEventos、eventEngine、eventEngineTopics等 9. **归纳总结**: - 生成内容时需要带简单的排版格式

正文

事件驱动模式

举个例子🌰

大部分软件或者APP都有会有会员系统,当我们注册为会员时,商家一般会把我们拉入会员群、给我们发优惠券、推送欢迎语什么的。
image
值得注意的是:

  1. 注册成功后才会产生后面的这些动作;
  2. 注册成功后的这些动作没有先后执行顺序之分;
  3. 注册成功后的这些动作的执行结果不能互相影响;

传统写法

public Boolean doRegisterVip(){
	//1、注册会员
	registerVip();
	//2、入会员群
	joinMembershipGroup();
	//3、发优惠券
	issueCoupons();
	//4、推送消息
	sendWelcomeMsg();
}

这样的写法将所有的动作都耦合在doRegisterVip方法中,首先执行效率低下,其次耦合度太高,最后不好扩展。那么如何优化这种逻辑呢?

事件驱动模式原理介绍🍓

Spring的事件驱动模型由三部分组成:

事件:用户可自定义事件类和相关属性及行为来表述事件特征,Spring4.2之后定义事件不需要再显式继承ApplicationEvent类,直接定义一个bean即可,Spring会自动通过PayloadApplicationEvent来包装事件。

事件发布者:在Spring中可通过ApplicationEventPublisher把事件发布出去,这样事件内容就可以被监听者消费处理。

事件监听者:ApplicationListener,监听发布事件,处理事件发生之后的后续操作。

原理图如下:
image

代码实现

1. 定义基本元素

事件发布者:EventEngine.java、EventEngineImpl.java

package com.example.event.config;

/**
 * 事件引擎
 */
public interface EventEngine {

    /**
     * 发送事件
     *
     * @param event 事件
     */
    void publishEvent(BizEvent event);
}
package com.example.event.config;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;

import org.springframework.util.CollectionUtils;

/**
 * 事件引擎实现类
 */
public class EventEngineImpl implements EventEngine {

    /**
     * 异步执行器。也系统需要自行定义线程池
     */
    private Executor bizListenerExecutor;

    /**
     * 是否异步,默认为false
     */
    private boolean async;

    /**
     * 订阅端 KEY是TOPIC,VALUES是监听器集合
     */
    private Map<String, List<BizEventListener>> bizSubscribers = new HashMap<>(16);

    @Override
    public void publishEvent(BizEvent event) {
        List<BizEventListener> listeners = bizSubscribers.get(event.getTopic());
        if (CollectionUtils.isEmpty(listeners)) {
            return;
        }
        for (BizEventListener bizEventListener : listeners) {
            if (bizEventListener.decide(event)) {
                //异步执行的话,放入线程池
                if (async) {
                    bizListenerExecutor.execute(new EventSubscriber(bizEventListener, event));
                } else {
                    bizEventListener.onEvent(event);
                }

            }
        }
    }

    /**
     * Setter method for property <tt>bizListenerExecutor</tt>.
     *
     * @param bizListenerExecutor value to be assigned to property bizListenerExecutor
     */
    public void setBizListenerExecutor(Executor bizListenerExecutor) {
        this.bizListenerExecutor = bizListenerExecutor;
    }

    /**
     * Setter method for property <tt>bizSubscribers</tt>.
     *
     * @param bizSubscribers value to be assigned to property bizSubscribers
     */
    public void setBizSubscribers(Map<String, List<BizEventListener>> bizSubscribers) {
        this.bizSubscribers = bizSubscribers;
    }

    /**
     * Setter method for property <tt>async</tt>.
     *
     * @param async value to be assigned to property async
     */
    public void setAsync(boolean async) {
        this.async = async;
    }
}

事件:BizEvent.java

package com.example.event.config;

import java.util.EventObject;

/**
 * 业务事件
 */
public class BizEvent extends EventObject {

    /**
     * Topic
     */
    private final String topic;

    /**
     * 业务id
     */
    private final String bizId;

    /**
     * 数据
     */
    private final Object data;

    /**
     * @param topic 事件topic,用于区分事件类型
     * @param bizId 业务ID,标识这一次的调用
     * @param data  事件传输对象
     */
    public BizEvent(String topic, String bizId, Object data) {
        super(data);
        this.topic = topic;
        this.bizId = bizId;
        this.data = data;
    }

    /**
     * Getter method for property <tt>topic</tt>.
     *
     * @return property value of topic
     */
    public String getTopic() {
        return topic;
    }

    /**
     * Getter method for property <tt>id</tt>.
     *
     * @return property value of id
     */
    public String getBizId() {
        return bizId;
    }

    /**
     * Getter method for property <tt>data</tt>.
     *
     * @return property value of data
     */
    public Object getData() {
        return data;
    }
}

事件监听者:EventSubscriber.java

package com.example.event.config;

/**
 * 事件监听者。注意:此时已经没有线程上下文,如果需要请修改构造函数,显示复制上下文信息
 */
public class EventSubscriber implements Runnable {

    /**
     * 业务监听器
     **/
    private BizEventListener bizEventListener;

    /**
     * 业务事件
     */
    private BizEvent bizEvent;

    /**
     * @param bizEventListener 事件监听者
     * @param bizEvent         事件
     */
    public EventSubscriber(BizEventListener bizEventListener, BizEvent bizEvent) {
        super();
        this.bizEventListener = bizEventListener;
        this.bizEvent = bizEvent;
    }

    @Override
    public void run() {
        bizEventListener.onEvent(bizEvent);
    }
}

2. 其他组件

业务事件监听器:BizEventListener.java

package com.example.event.config;

import java.util.EventListener;

/**
 * 业务事件监听器
 *
 */
public interface BizEventListener extends EventListener {

    /**
     * 是否执行事件
     *
     * @param event 事件
     * @return
     */
    public boolean decide(BizEvent event);

    /**
     * 执行事件
     *
     * @param event 事件
     */
    public void onEvent(BizEvent event);
}

事件引擎topic:EventEngineTopic.java

package com.example.event.config;

/**
 * 事件引擎topic,用于区分事件类型
 */
public class EventEngineTopic {
    /**
     * 入会员群
     */
    public static final String JOIN_MEMBERSHIP_GROUP = "joinMembershipGroup";

    /**
     * 发优惠券
     */
    public static final String ISSUE_COUPONS = "issueCoupons";

    /**
     * 推送消息
     */
    public static final String SEND_WELCOME_MSG = "sendWelcomeMsg";

}

3. 监听器实现

优惠券处理器:CouponsHandlerListener.java

package com.example.event.listener;

import com.example.event.config.BizEvent;
import com.example.event.config.BizEventListener;
import org.springframework.stereotype.Component;

/**
 * 优惠券处理器
 */
@Component
public class CouponsHandlerListener implements BizEventListener {

    @Override
    public boolean decide(BizEvent event) {
        return true;
    }

    @Override
    public void onEvent(BizEvent event) {
        System.out.println("优惠券处理器:十折优惠券已发放");
    }
}

会员群处理器:MembershipHandlerListener.java

package com.example.event.listener;

import com.example.event.config.BizEvent;
import com.example.event.config.BizEventListener;
import org.springframework.stereotype.Component;

/**
 * 会员群处理器
 */
@Component
public class MembershipHandlerListener implements BizEventListener {
    @Override
    public boolean decide(BizEvent event) {
        return true;
    }

    @Override
    public void onEvent(BizEvent event) {
        System.out.println("会员群处理器:您已成功加入会员群");
    }
}

消息推送处理器:MsgHandlerListener.java

package com.example.event.listener;

import com.example.event.config.BizEvent;
import com.example.event.config.BizEventListener;
import org.springframework.stereotype.Component;

/**
 * 消息推送处理器
 */
@Component
public class MsgHandlerListener implements BizEventListener {

    @Override
    public boolean decide(BizEvent event) {
        return true;
    }

    @Override
    public void onEvent(BizEvent event) {
        System.out.println("消息推送处理器:欢迎成为会员!!!");
    }
}

事件驱动引擎配置:EventEngineConfig.java

package com.example.event.listener;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.example.event.config.BizEventListener;
import com.example.event.config.EventEngine;
import com.example.event.config.EventEngineImpl;
import com.example.event.config.EventEngineTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

/**
 * 事件驱动引擎配置
 */
@Configuration
public class EventEngineConfig {
    /**
     * 线程池异步处理事件
     */
    private static final Executor EXECUTOR = new ThreadPoolExecutor(20, 50, 10, TimeUnit.MINUTES,
        new LinkedBlockingQueue(500), new CustomizableThreadFactory("EVENT_ENGINE_POOL"));

    @Bean("eventEngineJob")
    public EventEngine initJobEngine(CouponsHandlerListener couponsHandlerListener,
        MembershipHandlerListener membershipHandlerListener,
        MsgHandlerListener msgHandlerListener) {
        Map<String, List<BizEventListener>> bizEvenListenerMap = new HashMap<>();
        //注册优惠券事件
        bizEvenListenerMap.put(EventEngineTopic.ISSUE_COUPONS, Arrays.asList(couponsHandlerListener));
        //注册会员群事件
        bizEvenListenerMap.put(EventEngineTopic.JOIN_MEMBERSHIP_GROUP, Arrays.asList(membershipHandlerListener));
        //注册消息推送事件
        bizEvenListenerMap.put(EventEngineTopic.SEND_WELCOME_MSG, Arrays.asList(msgHandlerListener));

        EventEngineImpl eventEngine = new EventEngineImpl();
        eventEngine.setBizSubscribers(bizEvenListenerMap);
        eventEngine.setAsync(true);
        eventEngine.setBizListenerExecutor(EXECUTOR);
        return eventEngine;
    }
}

4. 测试类

TestController.java

package com.example.event.controller;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

import javax.annotation.Resource;

import com.example.event.config.BizEvent;
import com.example.event.config.EventEngine;
import com.example.event.config.EventEngineTopic;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * 测试
 */
@RestController
@RequestMapping("/test")
public class TestController {

    @Resource(name = "eventEngineJob")
    private EventEngine eventEngine;

    @GetMapping("/doRegisterVip")
    public String doRegisterVip(@RequestParam(required = true) String userName,
        @RequestParam(required = true) Integer age) {
        Map<String, Object> paramMap = new HashMap<>(16);
        paramMap.put("userName", userName);
        paramMap.put("age", age);
        //1、注册会员,这里不实现了
        System.out.println("注册会员成功");
        //2、入会员群
        eventEngine.publishEvent(
            new BizEvent(EventEngineTopic.JOIN_MEMBERSHIP_GROUP, UUID.randomUUID().toString(), paramMap));
        //3、发优惠券
        eventEngine.publishEvent(
            new BizEvent(EventEngineTopic.ISSUE_COUPONS, UUID.randomUUID().toString(), paramMap));
        //4、推送消息
        eventEngine.publishEvent(
            new BizEvent(EventEngineTopic.SEND_WELCOME_MSG, UUID.randomUUID().toString(), paramMap));
        return "注册会员成功";
    }
}

项目代码结构

image

调用接口

http://localhost:8080/test/doRegisterVip?userName=zhangsan&age=28

image

与基于Spring事件驱动模式实现业务解耦相似的内容:

基于Spring事件驱动模式实现业务解耦

事件驱动模式 举个例子🌰 大部分软件或者APP都有会有会员系统,当我们注册为会员时,商家一般会把我们拉入会员群、给我们发优惠券、推送欢迎语什么的。 值得注意的是: 注册成功后才会产生后面的这些动作; 注册成功后的这些动作没有先后执行顺序之分; 注册成功后的这些动作的执行结果不能互相影响; 传统写法

Spring Reactor基本介绍和案例

1. Reactor 对比 1.1 Reactor 线程模型 Reactor 线程模型就是通过 单个线程 使用 Java NIO 包中的 Selector 的 select()方法,进行监听。当获取到事件(如 accept、read 等)后,就会分配(dispatch)事件进行相应的事件处理(han

利用SpringBoot项目做一个Mock挡板;基于事件发布动态自定义URL和响应报文

# 导入SpringbootWEb依赖 ```xml org.springframework.boot spring-boot-starter-web ${spring-boot-start-version} org.springframework.boot spring-boot-starter-

基于 Spring Cloud 的微服务脚手架

基于 Spring Cloud 的微服务脚手架 作者: Grey 原文地址: 博客园:基于 Spring Cloud 的微服务脚手架 CSDN:基于 Spring Cloud 的微服务脚手架 本文主要介绍了基于 Spring Cloud Finchley 和 Spring Boot 2.0.x 版本

基于Spring-AOP的自定义分片工具

作者:陈昌浩 1 背景 随着数据量的增长,发现系统在与其他系统交互时,批量接口会出现超时现象,发现原批量接口在实现时,没有做分片处理,当数据过大时或超过其他系统阈值时,就会出现错误。由于与其他系统交互比较多,一个一个接口做分片优化,改动量较大,所以考虑通过AOP解决此问题。 2 Spring-AOP

基于Spring Cache实现Caffeine、jimDB多级缓存实战

在早期参与涅槃氛围标签中台项目中,前台要求接口性能999要求50ms以下,通过设计Caffeine、ehcache堆外缓存、jimDB三级缓存,利用内存、堆外、jimDB缓存不同的特性提升接口性能, 内存缓存采用Caffeine缓存,利用W-TinyLFU算法获得更高的内存命中率;同时利用堆外缓存降低内存缓存大小,减少GC频率,同时也减少了网络IO带来的性能消耗;利用JimDB提升接口高可用、高并

基于Spring事务的可靠异步调用实践

SpringTxAsync组件是仓储平台组(WMS6)自主研发的一个专门用于解决可靠异步调用问题的组件。 通过使用SpringTxAsync组件,我们成功地解决了在仓储平台(WMS6)中的异步调用需求。经过近二年多的实践并经历了两次618活动以及两次双11活动,该组件已经在我们的所有应用中稳定运行并

聊聊Spring Cloud Gateway

Spring Cloud Gateway是基于Spring Boot 2.0、Spring WebFlux和Project Reactor等技术开发的网关,它不仅提供了统一的路由请求的方式,还基于过滤链的方式提供了网关最基本的功能;解决了Spring Cloud Zuul的性能问题。

聊聊Spring IOC容器的注入方式

为什么要说这个? 对于Spring体系而言,我个人认为最重要的就是IOC容器,其次才是AOP、Context等模块;因为这些模块功能是或搭建或集成在IOC容器这个基础设施之上的。 直接基于Spring框架体系做开发时,可以通过常用的JavaConfig或XML方式将对象的生命周期及装配由容器原生的接

Spring源码阅读系列--全局目录

> 阅读之前要注意的东西:本文就是主打流水账式的源码阅读,主导的是一个参考,主要内容需要看官自己去源码中验证。全系列文章基于 spring 源码 5.x 版本。 写在开始前的话: 阅读spring 源码实在是一件庞大的工作,不说全部内容,单就最基本核心部分包含的东西就需要很长时间去消化了: - be