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

概述

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

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

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


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

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

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

// 第一种实现方式,其他类代码不变,就没有再重复罗列
public class RegPromotionObserver implements RegObserver {private PromotionService promotionService; // 依赖注入@Overridepublic void handleRegSuccess(long userId) {Thread thread = new Thread(new Runnable() {@Overridepublic 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() {@Overridepublic 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; // 依赖注入@Subscribepublic void handleRegSuccess(long userId) {promotionService.issueNewUserExperienceCash(userId);}
}public class RegNotificationObserver {private NotificationService notificationService;@Subscribepublic 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 {// 省略其他属性和方法...@Subscribepublic void f1(PMsg event) { /*...*/ }@Subscribepublic 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() {@Overridepublic 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。在平时的业务开发中,我们要善于抽象这些非业务的、可复用的功能,并积极把它们实现成通用的框架。

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

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

最优算法100例之18-列升序行升序的数组中查找元素

专栏主页:计算机专业基础知识总结(适用于期末复习考研刷题求职面试)系列文章https://blog.csdn.net/seeker1994/category_12585732.html 题目描述 在一个二维数组中,每一行都按照从左到右递增的顺序排序,每一列都按照从上到下递增的顺序排序。请完成一个函数,输入这样一…

蓝桥杯22年第十三届省赛-统计子矩阵|一维前缀和加双指针

题目链接&#xff1a; 1.统计子矩阵 - 蓝桥云课 (lanqiao.cn) 蓝桥杯2022年第十三届省赛真题-统计子矩阵 - C语言网 (dotcpp.com) 说明&#xff1a; 涉及到子矩阵的时候&#xff0c;一般就跟前缀和相关&#xff0c;可以降维。 先回顾一下最大子矩阵&#xff0c;回忆一下一…

AI绘画教程:Midjourney使用方法与技巧从入门到精通

文章目录 一、《AI绘画教程&#xff1a;Midjourney使用方法与技巧从入门到精通》二、内容介绍三、作者介绍&#x1f324;️粉丝福利 一、《AI绘画教程&#xff1a;Midjourney使用方法与技巧从入门到精通》 一本书读懂Midjourney绘画&#xff0c;让创意更简单&#xff0c;让设计…

[图像处理] MFC载入图片并进行二值化处理和灰度处理及其效果显示

文章目录 工程效果重要代码完整代码参考 工程效果 载入图片&#xff0c;并在左侧显示原始图片、二值化图片和灰度图片。 双击左侧的图片控件&#xff0c;可以在右侧的大控件中&#xff0c;显示双击的图片。 初始画面&#xff1a; 载入图片&#xff1a; 双击左侧的第二个控件…

随便注【强网杯2019】

大佬的完整wp&#xff1a;buuctf-web-[强网杯 2019]随便注-wp_取材于某次真实环境渗透,只说一句话:开发和安全缺一不可-CSDN博客 知识点&#xff1a; 单引号字符型绕过堆叠注入 可以执行多条语句multi_query()&#xff1a;该函数可能引发堆叠注入handler用法 mysql专属&#…

Linux之进程间通信

1.进程间通信的目的 数据传输&#xff1a;一个进程需要将它的数据发送给另一个进程 资源共享&#xff1a;多个进程之间共享同样的资源。 通知事件&#xff1a;一个进程需要向另一个或一组进程发送消息&#xff0c;通知它&#xff08;它们&#xff09;发生了某种事件&#xff…

Rabbit简单模式理解

简单模式 我们以最普通的方式去理解&#xff0c;并没有整合Springboot的那种 这是最简单的模式&#xff0c;一个生产者&#xff0c;一个消费者&#xff0c;一个队列 测试 1、 导包&#xff0c;没整合&#xff0c;不需要编写配置 2、需要生产者消费者 导包 <dependency…

【Spring】通过Spring收集自定义注解标识的方法

文章目录 前言1. 声明注解2. 使用 Spring 的工厂拓展3. 收集策略4. 完整的代码后记 前言 需求&#xff1a; 用key找到对应的方法实现。使用注解的形式增量开发。 MyComponent public class Sample1 {MyMethod(key "key1")public String test2() {return "She…

【SpringCloud】Eureka注册中心

目 录 一.Eureka的结构和作用二.搭建 eureka-server1. 创建 eureka-server 服务2. 引入 eureka 依赖3. 编写启动类4. 编写配置文件5. 启动服务 三.服务注册1. 引入依赖2. 配置文件3. 启动多个user-service实例 四.服务发现1. 引入依赖2. 配置文件3. 服务拉取和负载均衡 总结 假…

Optimizer神经网络中各种优化器介绍

1. SGD 1.1 batch-GD 每次更新使用全部的样本&#xff0c;注意会对所有的样本取均值&#xff0c;这样每次更新的速度慢。计算量大。 1.2 SGD 每次随机取一个样本。这样更新速度更快。SGD算法在于每次只去拟合一个训练样本&#xff0c;这使得在梯度下降过程中不需去用所有训…

阿里云通用算力型u1云服务器配置性能评测及价格参考

阿里云服务器u1是通用算力型云服务器&#xff0c;CPU采用2.5 GHz主频的Intel(R) Xeon(R) Platinum处理器&#xff0c;ECS通用算力型u1云服务器不适用于游戏和高频交易等需要极致性能的应用场景及对业务性能一致性有强诉求的应用场景(比如业务HA场景主备机需要性能一致)&#xf…

C++经典面试题目(十六)

1、什么是类的构造函数&#xff1f;它有什么作用&#xff1f; 类的构造函数是一种特殊的成员函数&#xff0c;用于初始化类的对象。它在对象被创建时自动调用&#xff0c;可以用于执行对象的初始化工作&#xff0c;如分配内存、初始化成员变量等。构造函数的作用在于确保对象在…

002 HTML元素

文章目录 HTML的元素元素的属性 注释快捷键根元素head元素meta元素 body元素h元素p元素img元素src属性alt属性 路径分隔符a元素href属性target属性锚点链接 iframe元素div与span不常用元素HTML全局属性字符实体URL和URISEO字符编码 官方文档:https://www.w3.org/TR/ HTML的元素…

【数据结构】括号匹配问题你学会了吗?来刷刷题检验一下吧!!!

栈在括号问题中的应用 导言一、有效的括号——栈、字符串——简单1.1 题目要求与分析1.2 代码实现 二、 最长有效括号——栈、字符串、动态规划——困难2.1 题目要求与分析2.2 问题解析2.2.1 如何计算有效括号的个数2.2.2 如何记录了连续括号的长度2.2.3 如何寻找最长的子串 2.…

STM32 GPIO输入检测——按键

前言 在嵌入式系统开发中&#xff0c;对GPIO输入进行检测是一项常见且关键的任务。STM32微控制器作为一款功能强大的处理器&#xff0c;具有丰富的GPIO功能&#xff0c;可以轻松实现对外部信号的检测和处理。在本文中&#xff0c;我们将深入探讨如何在STM32微控制器上进行GPIO…

二十四种设计模式与六大设计原则(三):【装饰模式、迭代器模式、组合模式、观察者模式、责任链模式、访问者模式】的定义、举例说明、核心思想、适用场景和优缺点

接上次博客&#xff1a;二十四种设计模式与六大设计原则&#xff08;二&#xff09;&#xff1a;【门面模式、适配器模式、模板方法模式、建造者模式、桥梁模式、命令模式】的定义、举例说明、核心思想、适用场景和优缺点-CSDN博客 目录 装饰模式【Decorator Pattern】 定义…

linux redis 快速安装

选择合适的版本 以下是在Linux系统上安装Redis的步骤&#xff1a; 1、下载Redis&#xff1a; wget http://download.redis.io/releases/redis-7.2.0.tar.gz 2、解压缩Redis&#xff1a; tar xzf redis-7.2.0.tar.gz 3、编译Redis&#xff1a; cd redis-7.2.0 make 4、…

C语言——内存函数

前言&#xff1a; C语言中除了字符串函数和字符函数外&#xff0c;还有一些函数可以直接对内存进行操作&#xff0c;这些函数被称为内存函数&#xff0c;这些函数与字符串函数都属于<string.h>这个头文件中。 一.memcpy&#xff08;&#xff09;函数 memcpy是C语言中的…

维修贝加莱4PP420.1043-B5触摸屏Power Panel 400工业电脑液晶

深圳捷达工控维修为贝加莱、HMI 显示电源面板 400 4PP420.1043-B5 提供专业电子维修。在 深圳捷达工控维修&#xff0c;我们拥有及时且经济高效地维修 B&R 、HMI Display Power Panel 400 4PP420.1043-B5 的经验。我们为发送给我们工厂维修的贝加莱 HMI 显示面板 400 4PP42…

【智能算法】霜冰优化算法(RIME)原理及实现

目录 1.背景2.算法原理2.1算法思想2.2算法过程 3.结果展示4.参考文献 1.背景 2023年&#xff0c;H Su等人受到自然界霜冰生长机制启发&#xff0c;提出了霜冰优化算法&#xff08;Rime Optimization Algorithm, RIME&#xff09;。 2.算法原理 2.1算法思想 RIME模拟软霜颗粒…