设计模式学习笔记 - 设计模式与范式 -行为型: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 题目描述 在一个二维数组中,每一行都按照从左到右递增的顺序排序,每一列都按照从上到下递增的顺序排序。请完成一个函数,输入这样一…

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…

【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…

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

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

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

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

C语言——内存函数

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

【智能算法】霜冰优化算法(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模拟软霜颗粒…

【Laravel】06 数据库迁移工具migration

【Laravel】06 数据库迁移工具migration 1.migration文件目录2. 举例 1.migration文件目录 2. 举例 (base) ➜ example-app php artisan migrate Migration table created successfully. Migrating: 2014_10_12_000000_create_users_table Migrated: 2014_10_12_000000_crea…

HTTP/1.1、HTTP/2、HTTP/3 演变(计算机网络)

HTTP/1.1 相比 HTTP/1.0 提高了什么性能&#xff1f; HTTP/1.1 相比 HTTP/1.0 性能上的改进&#xff1a; 使用长连接改善了短连接造成的性能开销。支持管道网络传输&#xff0c;只要第一个请求发出去了&#xff0c;不必等其回来&#xff0c;就可以发第二个请求出去&#xff0c…

0基础学习Mybatis系列数据库操作框架——最小Demo

大纲 数据库Mybatis目录结构配置代码 代码/配置结构配置结构依赖库配置SQL配置Mybatis配置数据库配置SQL映射配置 完整配置 代码结构映射类Mybatis逻辑从 XML 中构建 SqlSessionFactory从 SqlSessionFactory 中获取 SqlSession通过mapper中的namespace和id执行SQL 完整逻辑和代…

nginx的https与动态负载均衡

nginx的https 证书可以根据你的域名和服务器服务商去进行签发 , 比如 : 阿里云 腾讯云 百度云 华为云等 这里使用的是腾讯云 : 下载证书 : 选择 nginx: 下载之后传递到服务器上。 下面开始配置nginx的https: 1. 解压下载的证书包 cd /etc/ssl unzip xxcc.dwa_nginx.zip mv…

基于知识图谱的个性化学习推荐系统的设计与实现(论文+源码)_kaic

摘 要 Abstract 1 绪 论 1.1 研究背景及意义 1.2 国内外现状研究 1.3 研究工作和论文结构 2 相关技术 2.1 HTML 语言 2.2 Python 语言 2.3 数据库技术 2.4 Django 框架 3 系统分析 3.1 需求概述 3.2 系统可行性分析 3.2.1 技术可行性 3.2.2 经济可行性 3.2.3 操作可行性 3.3 功…

EXCEL-VB编程实现自动抓取多工作簿多工作表中的单元格数据

一、VB编程基础 1、 EXCEL文件启动宏设置 文件-选项-信任中心-信任中心设置-宏设置-启用所有宏 汇总文件保存必须以宏启动工作簿格式类型进行保存 2、 VB编程界面与入门 参考收藏 https://blog.csdn.net/O_MMMM_O/article/details/107260402?spm1001.2014.3001.5506 二、…

HarmonyOS实战开发-slider组件的使用

介绍 本篇Codelab主要介绍slider滑动条组件的使用。如图所示拖动对应滑动条调节风车的旋转速度以及缩放比例。 相关概念 slider组件&#xff1a;滑动条组件&#xff0c;通常用于快速调节设置值&#xff0c;如音量调节、亮度调节等应用场景。 环境搭建 软件要求 DevEco Stu…