概述
《1.观察者模式(上)》我们学习了观察者模式的原理、实现、应用场景,重点节介绍了不同应用场景下,几种不同的实现方式,包括:同步阻塞、异步非阻塞、进程内、进程间的实现方式。
同步阻塞最经典的实现方式,主要是为了解耦;异步非阻塞除了能实现代码解耦之外,还能提高代码的执行效率;进程间的观察者模式解耦更加彻底,一般是基于消息队列来实现,用来实现不同进程间被观察者和观察者之间的交互。
今天,我们聚焦于异步非阻塞的观察者模式,带你实现一个类似 Google Guava EventBus 的通用框架。
异步非阻塞观察者模式的简易实现
对于异步非阻塞观察者模式,如果只是实现一个简易版本,不用考虑任何通用性、复杂性、复用性,实际上是非常容易的。
我们有两种实现方式。其中一种是:在每个 handleRegSuccess()
函数中创建一个新的线程执行代码逻辑;另一种是:在 UserController
的 register()
函数中使用线程池来执行每个观察者的 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); => AObserver、BObserver接收到消息
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 个类: 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() {
@Override
public 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。在平时的业务开发中,我们要善于抽象这些非业务的、可复用的功能,并积极把它们实现成通用的框架。