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

2024-04-01 06:52
文章标签 实现 学习 设计模式 模式 观察者 笔记 行为 框架 异步 阻塞 范式 eventbus

本文主要是介绍设计模式学习笔记 - 设计模式与范式 -行为型：2.观察者模式（下）：实现一个异步非阻塞的EventBus框架，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

概述

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

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

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

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

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

我们有两种实现方式。其中一种是：在每个 handleRegSuccess() 函数中创建一个新的线程执行代码逻辑；另一种是：在 UserController 的 register() 函数中使用线程池来执行每个观察者的 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); => AObserver、BObserver接收到消息
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 个类： EventBus、 AsyncEventBUs、Subscribe、ObserverAction、ObserverRegistry。接下来，依次来看下这 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

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

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

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

总结

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

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

这篇关于设计模式学习笔记 - 设计模式与范式 -行为型：2.观察者模式（下）：实现一个异步非阻塞的EventBus框架的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/866490。

相关文章

Java实现检查多个时间段是否有重合

Java实现检查多个时间段是否有重合

《Java实现检查多个时间段是否有重合》这篇文章主要为大家详细介绍了如何使用Java实现检查多个时间段是否有重合,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录流程概述步骤详解China编程步骤1：定义时间段类步骤2：添加时间段步骤3：检查时间段是否有重合步骤4：输出结果示例代码结语作
阅读更多...
使用C++实现链表元素的反转

使用C++实现链表元素的反转

《使用C++实现链表元素的反转》反转链表是链表操作中一个经典的问题,也是面试中常见的考题,本文将从思路到实现一步步地讲解如何实现链表的反转,帮助初学者理解这一操作,我们将使用C++代码演示具体实现,同... 目录问题定义思路分析代码实现带头节点的链表代码讲解其他实现方式时间和空间复杂度分析总结问题定义给定
阅读更多...
Java覆盖第三方jar包中的某一个类的实现方法

Java覆盖第三方jar包中的某一个类的实现方法

《Java覆盖第三方jar包中的某一个类的实现方法》在我们日常的开发中,经常需要使用第三方的jar包,有时候我们会发现第三方的jar包中的某一个类有问题,或者我们需要定制化修改其中的逻辑,那么应该如何... 目录一、需求描述二、示例描述三、操作步骤四、验证结果五、实现原理一、需求描述需求描述如下：需要在
阅读更多...
如何使用Java实现请求deepseek

如何使用Java实现请求deepseek

《如何使用Java实现请求deepseek》这篇文章主要为大家详细介绍了如何使用Java实现请求deepseek功能,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录1.deepseek的api创建2.Java实现请求deepseek2.1 pom文件2.2 json转化文件2.2
阅读更多...
python使用fastapi实现多语言国际化的操作指南

python使用fastapi实现多语言国际化的操作指南

《python使用fastapi实现多语言国际化的操作指南》本文介绍了使用Python和FastAPI实现多语言国际化的操作指南,包括多语言架构技术栈、翻译管理、前端本地化、语言切换机制以及常见陷阱和... 目录多语言国际化实现指南项目多语言架构技术栈目录结构翻译工作流1. 翻译数据存储2. 翻译生成脚本
阅读更多...
如何通过Python实现一个消息队列

如何通过Python实现一个消息队列

《如何通过Python实现一个消息队列》这篇文章主要为大家详细介绍了如何通过Python实现一个简单的消息队列,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录如何通过 python 实现消息队列如何把 http 请求放在队列中执行1. 使用 queue.Queue 和 reque
阅读更多...
Python如何实现PDF隐私信息检测

Python如何实现PDF隐私信息检测

《Python如何实现PDF隐私信息检测》随着越来越多的个人信息以电子形式存储和传输,确保这些信息的安全至关重要,本文将介绍如何使用Python检测PDF文件中的隐私信息,需要的可以参考下... 目录项目背景技术栈代码解析功能说明运行结php果在当今，数据隐私保护变得尤为重要。随着越来越多的个人信息以电子形
阅读更多...
使用 sql-research-assistant进行 SQL 数据库研究的实战指南(代码实现演示)

使用 sql-research-assistant进行 SQL 数据库研究的实战指南(代码实现演示)

《使用sql-research-assistant进行SQL数据库研究的实战指南(代码实现演示)》本文介绍了sql-research-assistant工具,该工具基于LangChain框架,集... 目录技术背景介绍核心原理解析代码实现演示安装和配置项目集成LangSmith 配置（可选）启动服务应用场景
阅读更多...
使用Python快速实现链接转word文档

使用Python快速实现链接转word文档

《使用Python快速实现链接转word文档》这篇文章主要为大家详细介绍了如何使用Python快速实现链接转word文档功能,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 演示代码展示from newspaper import Articlefrom docx import
阅读更多...
前端原生js实现拖拽排课效果实例

前端原生js实现拖拽排课效果实例

《前端原生js实现拖拽排课效果实例》：本文主要介绍如何实现一个简单的课程表拖拽功能,通过HTML、CSS和JavaScript的配合,我们实现了课程项的拖拽、放置和显示功能,文中通过实例代码介绍的... 目录1. 效果展示2. 效果分析2.1 关键点2.2 实现方法3. 代码实现3.1 html部分3.2
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2