设计模式学习笔记 - 设计模式与范式 -行为型:2.观察者模式(下):实现一个异步非阻塞的EventBus框架

概述

《1.观察者模式(上)》我们学习了观察者模式的原理、实现、应用场景,重点节介绍了不同应用场景下,几种不同的实现方式,包括:同步阻塞、异步非阻塞、进程内、进程间的实现方式。

同步阻塞最经典的实现方式,主要是为了解耦;异步非阻塞除了能实现代码解耦之外,还能提高代码的执行效率;进程间的观察者模式解耦更加彻底,一般是基于消息队列来实现,用来实现不同进程间被观察者和观察者之间的交互。

今天,我们聚焦于异步非阻塞的观察者模式,带你实现一个类似 Google Guava EventBus 的通用框架。


异步非阻塞观察者模式的简易实现

对于异步非阻塞观察者模式,如果只是实现一个简易版本,不用考虑任何通用性、复杂性、复用性,实际上是非常容易的。

我们有两种实现方式。其中一种是:在每个 handleRegSuccess() 函数中创建一个新的线程执行代码逻辑;另一种是:在 UserControllerregister() 函数中使用线程池来执行每个观察者的 handleRegSuccess() 函数。两种实现方式的具体代码如下所示:

// 第一种实现方式,其他类代码不变,就没有再重复罗列
public class RegPromotionObserver implements RegObserver {
    private PromotionService promotionService; // 依赖注入
    @Override
    public void handleRegSuccess(long userId) {
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                promotionService.issueNewUserExperienceCash(userId);
            }
        });
        thread.start();
    }
}

// 第二种实现方式,其他类代码不变,就没有再重复罗列
public class UserController {
    private List<RegObserver> regObservers = new ArrayList<>();
    private UserService userService; // 依赖注入
    private Executor executor;

    public UserController(Executor executor) {
        this.executor = executor;
    }

    // 一次性设置好,之后也不可能动态地修改
    public void setRegObservers(List<RegObserver> observers) {
        regObservers.addAll(observers);
    }

    public long register(String telephone, String password) {
        // 省略输入参数的校验代码
        // 省略userService.register()异常的try-catch代码
        long userId = userService.register(telephone, password);

        for (RegObserver observer : regObservers) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    observer.handleRegSuccess(userId);
                }
            });
        }
        return userId;
    }
}

对于第一种实现方式,频繁地创建和销毁线程比较好使,并且并发线程数无法控制,创建过多的线程会导致堆栈溢出。第二种实现方式,尽管利用了线程池解决了第一种实现方式的问题,但线程池、异步执行逻辑都耦合在了 register() 函数中,增加了这部分业务代码的维护成本。

如果我们的需求更加极端一点,需要在同步阻塞和异步非阻塞之间灵活切换,那就要不停地修改 UserController 的代码。此外,如果在项目中,不止一个业务模块需要用到异步非阻塞观察者模式,那这样的代码实现也无法做到复用。

我们知道,框架的作用有:隐藏细节,降低开发难度,做到代码复用,解耦业务与非业务代码,让程序员聚焦业务开发。针对异步非阻塞观察者模式,我们也可以将它抽象成框架来达到这样的效果,而这个框架就是我们这节课要将的 EventBus。

EventBus 框架功能需求介绍

EventBus 翻译为 “事件总线”,它提供了实现观察者模式的骨架代码。我们可以基于此框架,非常容易地在自己的业务场景中实现观察者模式,不需要从零开始开发。其中,Google Guava EventBus 是一个比较著名的 EventBus 框架,它不仅仅支持异步非阻塞模式,同时也支持同步阻塞模式。

现在,我们就通过例子来看一下,Guava EventBus 具有哪些工能够。还是《1.观察者模式(上)》中的例子,我们用 Guava EventBus 重新实现一下,代码如下所示:

public class UserController {
    private UserService userService; // 依赖注入

    private EventBus eventBus;
    private static final int DEFAULT_EVENTBUS_THREAD_POOL_SIZE = 20;

    public UserController() {
//        eventBus = new EventBus(); // 同步阻塞模式
        eventBus = new AsyncEventBus(Executors.newFixedThreadPool(
                DEFAULT_EVENTBUS_THREAD_POOL_SIZE)); // 异步非阻塞模式
    }

    // 一次性设置好,之后也不可能动态地修改
    public void setRegObservers(List<RegObserver> observers) {
        for (RegObserver observer : observers) {
            eventBus.register(observers);
        }
    }

    public long register(String telephone, String password) {
        // 省略输入参数的校验代码
        // 省略userService.register()异常的try-catch代码
        long userId = userService.register(telephone, password);

        eventBus.post(userId);

        return userId;
    }
}

public class RegPromotionObserver {
    private PromotionService promotionService; // 依赖注入

    @Subscribe
    public void handleRegSuccess(long userId) {
        promotionService.issueNewUserExperienceCash(userId);
    }
}

public class RegNotificationObserver {
    private NotificationService notificationService;
    @Subscribe
    public void handleRegSuccess(long userId) {
        notificationService.senInboxMessage(userId, "Welcome ...");
    }
}

利用 EventBus 框架实现的观察者模式,跟从零开始编写的观察者模式相比,从大的流程上来说,实现思路大致一样,都需要定义 Observer,并通过 register() 函数注册 Observer,也需要通过某个函数(比如 EventBus 中的 post())来给 Observer 发送消息(在 EventBus 中消息被称作事件 event)。

但在实现细节方面,它们又有些区别。基于 EventBus,我们不需要定义 Observer 接口,任意类型的对象都可以注册到 EventBus 中,通过 @Subscribe 注解来表明类中哪个函数可以接收被观察者发送的消息。

接下来,我们详细地讲一下,Guava EventBus 的几个主要的类和函数。

- EventBus、AsyncEventBus

Guava EventBus 对外暴露的所有可调用接口,都封装在 EventBus 类中。其中,EventBus 实现了同步阻塞的观察者模式,AsyncEventBus 继承自 EventBus,提供了异步非阻塞的观察者模式。具体使用方式如下所示:

eventBus = new EventBus(); // 同步阻塞模式
eventBus = new AsyncEventBus(Executors.newFixedThreadPool(8)); // 异步非阻塞模式

- register()函数

EventBus 提供了 register() 函数用来注册观察者。具体的函数定义如下所示。它可以接受任何类型(Object)的观察者。而在经典的观察者模式的实现中,register() 函数必须接受实现了同一 Observer 接口的类对象。

public void register(Object object);

- unregister()函数

相对于 register() 函数,unregister() 函数用来从 EventBus 中删除某个观察者。具体的函数定义如下所示:

public void unregister(Object object);

- post()函数

EventBus 提供了 post() 函数用来,用来给观察者发送消息。具体的函数定义如下所示:

public void post(Object event);

跟经典的观察者模式的不同之处在于,当我们调用 post() 函数发送消息的时候,并非把消息发送给所有的观察者,而是发送给可匹配的观察者。所谓可匹配指的是,能接收的消息类型是发送消息(post 函数中定义的 event)类型及其父类。举个例子来解释下。

比如 AObserver 能接收的消息类型是 XMsg,BObserver 能接收的消息类型是 YMsg,CObserver 能接收的消息类型是 ZMsg。其中,XMsg 是 YMsg 的父类。当我们如下发送消息时,相应能接收到消息的可匹配观察者如下所示:

XMsg xMsg = new XMsg();
YMsg yMsg = new YMsg();
ZMsg zMsg = new ZMsg();
post(xMsg); => AObserver接收到消息
post(yMsg); => AObserverBObserver接收到消息
post(zMsg); => CObserver接收到消息

你可能会问,每个 Observer 能接收的消息类型是在哪里定义的?我们来看下 Guava EventBus 最特别的一个地方,那就是 @Subscribe 注解。

- @Subscribe 注解

EventBus 通过 @Subscribe 注解来标明,某个函数能接收哪种类型的消息。具体的使用代码如下所示。在 DObserver 类中,我们通过 @Subscribe 注解了两个函数 f1()f2()

public DObserver {
	// 省略其他属性和方法...
	
	@Subscribe
	public void f1(PMsg event) { /*...*/ }
	
	@Subscribe
	public void f2(QMsg event) { /*...*/ }
}

当通过 register() 函数将 DObserver 类对象注册到 EventBus 的时候,EventBus 会根据 @Subscribe 注解找到 f1()f2(),并且将两个函数能接收的消息类型记录下来(PMsg->f1,QMsg->f2)。当我们通过 post() 函数发送消息(比如 QMsg)的时候,EventBus 会通过之前记录的 (QMsg->f2),调用响应的函数 f2()

手把手实现一个 EventBus 框架

Guava EventBus 的功能我们已经讲清楚了,总体上来说还是比较简单的。接下来,我们就重复造轮子,自己造一个 “EventBus” 出来。

我们重点来看下, EventBus 中两个核心函数 register()post() 的实现原理。弄懂了它们,基本上就弄懂了整个 EventBus 框架。下面两张图是这两个函数的实现原理图。

在这里插入图片描述
在这里插入图片描述
从图中可以看出,最关键的一个数据结构是 Observer 注册表,记录了消息类型和可接收消息函数的对应关系。

  • 当调用 register() 函数注册观察者的时候,EventBus 通过解析 @Subscribe 注解,生成 Observer 注册表。
  • 当调用 post() 函数发送消息的时候,EventBus 通过注册表找到相应的可接收消息的函数,然后通过 Java 的反射语法动态地创建对象、执行函数。对于同步阻塞模式,EventBus 在一个线程内依次执行相应的函数。对于异步非阻塞模式,EventBus 通过一个线程池来执行相应的函数。

弄懂了原理,实现起来就简单多了。整个小框架的代码实现包括 5 个类: EventBusAsyncEventBUsSubscribeObserverActionObserverRegistry。接下来,依次来看下这 5 个类。

1.Subscribe

Subscribe 是一个注解,用于标明观察者中的哪个函数可以接收消息。

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
@Beta
public @interface Subscribe {}

2.ObserverAction

ObserverAction 用来表示 @Subscribe 注解的方法,其中 target 表示观察者类,method 表示方法。它主要用在 ObserverRegistry 观察者注册表中。

public class ObserverAction {
    private Object target;
    private Method method;

    public ObserverAction(Object target, Method method) {
        this.target = target;
        this.method = method;
    }
    
    public void execute(Object event) { // event是method方法的参数
        try {
            method.invoke(target, event);
        } catch (IllegalAccessException | InvocationTargetException e) {
            e.printStackTrace();
        }
    }
}

3.ObserverRegistry

ObserverRegistry 就是 Observer 注册表,是最复杂的一个类,框架中几乎所有的核心逻辑都在这个类中。这个类大量使用了 Java 的反射语法,不过代码整体来说都不难理解,其中,一个比较有技巧的地方是 CopyOnWriteArraySet 的使用。

CopyOnWriteArraySet,顾名思义,在写入数据的时候,会创建一个新的 set,并且将原始数据 clone 到新的 set 中,在新的 set 中写入数据完成之后,再用新的 set 替换老的 set。这样就能保证在写入数据的时候,不影响数据的读取操作,以此来解决读写并发问题。此外,CopyOnWriteArraySet 还通过加锁的方式,避免了并发写冲突。具体的作用你可以去查看一下 CopyOnWriteArraySet 类的源码。

public class ObserverRegistry {
    private ConcurrentHashMap<Class<?>, CopyOnWriteArraySet<ObserverAction>> registry
            = new ConcurrentHashMap<>();

    public void register(Object observer) {
        Map<Class<?>, Collection<ObserverAction>> observerActions = findAllObserverActions(observer);
        for (Map.Entry<Class<?>, Collection<ObserverAction>> entry : observerActions.entrySet()) {
            Class<?> eventType = entry.getKey();
            Collection<ObserverAction> eventActions = entry.getValue();
            CopyOnWriteArraySet<ObserverAction> registeredEventActions = registry.get(eventType);
            if (registeredEventActions == null) {
                registry.putIfAbsent(eventType, new CopyOnWriteArraySet<>());
                registeredEventActions = registry.get(eventType);
            }
            registeredEventActions.addAll(eventActions);
        }
    }

    public List<ObserverAction> getMatchedObserverActions(Object event) {
        List<ObserverAction> matchedObserverActions = new ArrayList<>();
        Class<?> postedEventType = event.getClass();
        for (Map.Entry<Class<?>, CopyOnWriteArraySet<ObserverAction>> entry : registry.entrySet()) {
            Class<?> eventType = entry.getKey();
            Collection<ObserverAction> eventActions = entry.getValue();
            if (postedEventType.isAssignableFrom(eventType)) {
                matchedObserverActions.addAll(eventActions);
            }
        }
        return matchedObserverActions;
    }

    private Map<Class<?>, Collection<ObserverAction>> findAllObserverActions(Object observer) {
        Map<Class<?>, Collection<ObserverAction>> observerActions = new HashMap<>();
        Class<?> clazz = observer.getClass();
        for (Method method : getAnnotatedMethods(clazz)) {
            Class<?>[] parameterTypes = method.getParameterTypes();
            Class<?> eventType = parameterTypes[0];
            if (!observerActions.containsKey(eventType)) {
                observerActions.put(eventType, new ArrayList<>());
            }
            observerActions.get(eventType).add(new ObserverAction(observer, method));
        }
        return observerActions;
    }

    private List<Method> getAnnotatedMethods(Class<?> clazz) {
        List<Method> annotatedMethods = new ArrayList<>();
        for (Method method : clazz.getDeclaredMethods()) {
            if (method.isAnnotationPresent(Subscribe.class)) {
                Class<?>[] parameterTypes = method.getParameterTypes();
                Preconditions.checkArgument(parameterTypes.length == 1,
                        "Method %s has @Subscribe annotation but has %s parameters." +
                                "Subscriber methods must hava exactly 1 parameters",
                        method, parameterTypes.length);
                annotatedMethods.add(method);
            }
        }
        return annotatedMethods;
    }
}

4.EventBus

EventBus 实现的是同步阻塞观察者模式。MoreExecutors.directExecutor() 是 Google Guava 提供的工具类,看似是多线程,实际上是单线程。之所以这么实现,主要还是为了跟 AsyncEventBus 统一代码逻辑,做到代码复用。

public class EventBus {
    private Executor executor;
    private ObserverRegistry registry = new ObserverRegistry();

    public EventBus() {
        this(MoreExecutors.directExecutor());
    }

    public EventBus(Executor executor) {
        this.executor = executor;
    }
    
    public void register(Object object) {
        registry.register(object);
    }
    
    public void post(Object event) {
        List<ObserverAction> observerActions = registry.getMatchedObserverActions(event);
        for (ObserverAction observerAction : observerActions) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    observerAction.execute(event);
                }
            });
        }
    }
}

5.AsyncEventBus

有了 EventBusAsyncEventBus 的实现就非常简单了。为了实现异步非阻塞的观察者模式,它就不能再继续使用 MoreExecutors.directExecutor() 了,而是需要在构造函数中,由调用者注入线程池。

public class AsyncEventBus extends EventBus {
    public AsyncEventBus(Executor executor) {
        super(executor);
    }
}

至此,用了不到 200 行代码,就实现了一个还算凑活能用的 EventBus,从功能上来讲,它跟 Google Guava EventBus 几乎一样。不过,在细节方面,相比我们的实现,Google Guava EventBus 做了很多优化,比如优化了注册表中查找消息可匹配函数的算法。

总结

框架的作用有:隐藏实现细节,降低开发难度,做到代码复用,解耦业务与非业务代码,让程序员聚焦业务开发。针对异步非阻塞观察者模式,我们可以将它抽象成框架来达到这样的效果,而这个框架就是 EventBus。EventBus 被翻译为 “事件总线”,它提供了实现观察者模式的骨架代码。我们可以基于此框架,非常容易地在自己的业务场景中实现观察者模式,不需要从零开发。

很多人觉得做业务开发没有技术调整,实际上,做业务开发也会涉及很多非业务功能的开发,比如本章讲到的 EventBus。在平时的业务开发中,我们要善于抽象这些非业务的、可复用的功能,并积极把它们实现成通用的框架。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/506236.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

springboot配置文件application.properties,application.yml/application.yaml

application.properties Springboot提供的一种属性配置方式&#xff1a;application.properties 初始时配置文件中只有一行语句。 启动时&#xff0c;比如tomcat的端口号默认8080&#xff0c;路径默认。如果我们要改变端口号及路径之类的可以在application.properties中配置。…

基于微信小程序的自习室预约系统的设计与实现

基于微信小程序的自习室预约系统的设计与实现 文章目录 基于微信小程序的自习室预约系统的设计与实现1、前言介绍2、功能设计3、功能实现4、开发技术简介5、系统物理架构6、系统流程图7、库表设计8、关键代码9、源码获取10、 &#x1f389;写在最后 1、前言介绍 伴随着信息技术…

ESP8266 WiFi物联网智能插座—上位机软件实现

1、软件架构 上位机主要作为下位机数据上传服务端以及节点调试的控制端&#xff0c;可以等效认为是专属版本调试工具。针对智能插座协议&#xff0c;对于下位机进行可视化监测和管理。 软件技术架构如下&#xff0c;主要为针对 Windows 的PC 端应用程序&#xff0c;采用WPF以及…

pyqt 创建右键菜单栏

class MainModule(QMainWindow, Ui_MainWindow):def __init__(self):super().__init__(parentNone)self.setupUi(self)# 允许出现菜单栏self.tableWidget.setContextMenuPolicy(Qt.CustomContextMenu)# 对空间添加右键菜单栏处理 self.tableWidget.customContextMenuRequested.…

C练习题(1)

变种水仙花&#xff08;来自牛课网&#xff09; 题目 变种水仙花数 - Lily Number&#xff1a;把任意的数字&#xff0c;从中间拆分成两个数字&#xff0c;比如1461 可以拆分成&#xff08;1和461&#xff09;,&#xff08;14和61&#xff09;,&#xff08;146和1),如果所有拆…

【Web】NSSCTF Round#20 Basic 两道0解题的赛后谈

目录 前言 baby-Codeigniter 组合拳&#xff01; 前言 本想着说看看go的gin框架就睡了的&#xff0c;r3师傅提醒说赛题环境已经上了&#xff0c;那不赶紧研究下&#x1f600; 主要来谈谈做题的心路历程 baby-Codeigniter 拿到题目的第一反应应该是&#xff1a;“什么是C…

[ESP32]:基于esp-modbus实现serial主机

[ESP32]&#xff1a;基于esp-modbus实现serial主机 开发环境 esp idf 5.1esp-modbus 1.0.13 使用如下指令添加组件&#xff0c;或者访问esp-modbus idf.py add-dependency "espressif/esp-modbus^1.0.13"Device parameters 对于master而言&#xff0c;需要理解的…

五、Yocto集成QT5(基于Raspberrypi 4B)

Yocto集成QT5 本篇文章为基于raspberrypi 4B单板的yocto实战系列的第五篇文章&#xff1a; 一、yocto 编译raspberrypi 4B并启动 二、yocto 集成ros2(基于raspberrypi 4B) 三、Yocto创建自定义的layer和image 四、Yocto创建静态IP和VLAN 本章节实操代码请查看github仓库&…

CVAE——生成0-9数字图像(Pytorch+mnist)

1、简介 CVAE&#xff08;Conditional Variational Autoencoder&#xff0c;条件变分自编码器&#xff09;是一种变分自编码器&#xff08;VAE&#xff09;的变体&#xff0c;用于生成有条件的数据。在传统的变分自编码器中&#xff0c;生成的数据是完全由潜在变量决定的&…

GridLayoutManager 中的一些坑

前言 如果GridLayoutManager使用item的布局都是wrap_cotent 那么会在布局更改时会出现一些出人意料的情况。&#xff08;本文完全不具备可读性和说教性&#xff0c;仅为博主方便查找问题&#xff09; 布局item: <!--layout_item.xml--> <?xml version"1.0&qu…

论文阅读: Visual Attention Network

Motivation 自注意力机制在2D自然图像领域面临3个挑战&#xff1a; 视二维图像为一维序列。对于高分辨率图像&#xff0c;二次复杂度消耗太大。只捕捉空间适应性&#xff0c;忽略通道适应性。 Contribution 设计了 Large Kernel attention(LKA)&#xff0c;包含卷积和自注意…

SpringBoot整合knife4J 3.0.3

Knife4j的前身是swagger-bootstrap-ui,前身swagger-bootstrap-ui是一个纯swagger-ui的ui皮肤项目。项目正式更名为knife4j,取名knife4j是希望她能像一把匕首一样小巧,轻量,并且功能强悍,更名也是希望把她做成一个为Swagger接口文档服务的通用性解决方案,不仅仅只是专注于前端Ui…

受益于边缘计算的三个关键应用

边缘计算和 5G 网络正在改变物联网&#xff0c;增强跨多个领域的广泛应用的功能&#xff0c;并催生大量新兴应用。我们通过研究三个突出的用例来说明边缘计算的强大功能。 工业4.0智能工厂 工业 4.0 为制造商提供了基于灵活的工业环境提高生产力和盈利能力的愿景&#xff0c;…

5.vector容器的使用

文章目录 vector容器1.构造函数代码工程运行结果 2.赋值代码工程运行结果 3.容量和大小代码工程运行结果 4.插入和删除代码工程运行结果 5.数据存取工程代码运行结果 6.互换容器代码工程运行结果 7.预留空间代码工程运行结果 vector容器 1.构造函数 /*1.默认构造-无参构造*/ …

STM32 can通信部分函数注释

相关截图: CAN模式初始化函数:u8 CAN1_Mode_Init(u8 tsjw,u8 tbs2,u8 tbs1,u16 brp,u8 mode) //CAN初始化 //tsjw:重新同步跳跃时间单元.范围:CAN_SJW_1tq~ CAN_SJW_4tq //tbs2:时间段2的时间单元. 范围:CAN_BS2_1tq~CAN_BS2_8tq; //tbs1:时间段1的时间单元. 范围:CAN_BS…

IO流c++

IO流类库 输入输出流 #include <iostream> using namespace std;class InCount { public:InCount(int a 0, int b 0){c1 a;c2 b;}void show(void){cout << "c1" << c1 << "\t" << "c2" << c2 << …

PHP三种方式读取RSA密钥加解密、签名验签完整教程

目录 第一步、生成公私钥 第二步、三种方式读取RSA密钥 第1种&#xff1a;公私钥弄成一行&#xff0c;必须一行没有空格和换行 第2种&#xff1a;直接复制生成公私钥 第3种;复制密钥存储为.pem文件后缀 第三步、RSA加解密 第四步、RSA签名以及验证签名 第五步、封装完整…

Linux的开发工具(二):编译器gcc/g++与Linux项目自动化构建工具-Makefile

Linux的编译器-gcc/g 基本概念&#xff1a;gcc是专门用来编译c语言的&#xff0c;g可以编译c或c语言 问题一&#xff1a;gcc有时候为什么不能编译带有for循环的c语言源文件&#xff1f; 答&#xff1a;gcc版本过低会不支持for循环等c99标准下的内容 解决方式&#xff1a;gcc…

手搓 Docker Image Creator(DIC)工具(02):预备知识

此节主要简单介绍一下 Docker、Dockerfile 的基本概念&#xff0c;Dockerfile 对的基本语法&#xff0c;Windows 和 macOS 下 Docker 桌面的安装&#xff0c;Docker 镜像的创建和运行测试等。 1 关于 Docker Docker 是一个开源的应用容器引擎&#xff0c;它允许开发者打包应用…

Open3D(C++) 基于随机抽样与特征值法的点云平面稳健拟合方法

目录 一、算法原理1、论文概述2、参考文献二、代码实现三、结果展示四、测试数据本文由CSDN点云侠原创,原文链接。如果你不是在点云侠的博客中看到该文章,那么此处便是不要脸的GPT爬虫。 一、算法原理 1、论文概述 针对点云数据含有异常值且传统拟合方法拟合结果不理想的情…