怎么看网站用的什么cms,西昌手机网站制作,知名的网页制作公司哪家好,wordpress5.2.2中文概述
《1.观察者模式#xff08;上#xff09;》我们学习了观察者模式的原理、实现、应用场景#xff0c;重点节介绍了不同应用场景下#xff0c;几种不同的实现方式#xff0c;包括#xff1a;同步阻塞、异步非阻塞、进程内、进程间的实现方式。
同步阻塞最经典的实现…概述
《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 ListRegObserver regObservers new ArrayList();private UserService userService; // 依赖注入private Executor executor;public UserController(Executor executor) {this.executor executor;}// 一次性设置好之后也不可能动态地修改public void setRegObservers(ListRegObserver 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(ListRegObserver 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 能接收的消息类型是 XMsgBObserver 能接收的消息类型是 YMsgCObserver 能接收的消息类型是 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-f1QMsg-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 ConcurrentHashMapClass?, CopyOnWriteArraySetObserverAction registry new ConcurrentHashMap();public void register(Object observer) {MapClass?, CollectionObserverAction observerActions findAllObserverActions(observer);for (Map.EntryClass?, CollectionObserverAction entry : observerActions.entrySet()) {Class? eventType entry.getKey();CollectionObserverAction eventActions entry.getValue();CopyOnWriteArraySetObserverAction registeredEventActions registry.get(eventType);if (registeredEventActions null) {registry.putIfAbsent(eventType, new CopyOnWriteArraySet());registeredEventActions registry.get(eventType);}registeredEventActions.addAll(eventActions);}}public ListObserverAction getMatchedObserverActions(Object event) {ListObserverAction matchedObserverActions new ArrayList();Class? postedEventType event.getClass();for (Map.EntryClass?, CopyOnWriteArraySetObserverAction entry : registry.entrySet()) {Class? eventType entry.getKey();CollectionObserverAction eventActions entry.getValue();if (postedEventType.isAssignableFrom(eventType)) {matchedObserverActions.addAll(eventActions);}}return matchedObserverActions;}private MapClass?, CollectionObserverAction findAllObserverActions(Object observer) {MapClass?, CollectionObserverAction 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 ListMethod getAnnotatedMethods(Class? clazz) {ListMethod 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) {ListObserverAction 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。在平时的业务开发中我们要善于抽象这些非业务的、可复用的功能并积极把它们实现成通用的框架。
