做淘宝站外推广网站,郑州加盟做网站,wordpress 页面空白页,建设网站的教程上一节课中#xff0c;我们学习了观察者模式的原理、实现、应用场景#xff0c;重点介绍了不同应用场景下#xff0c;几种不同的实现方式#xff0c;包括#xff1a;同步阻塞、异步非阻塞、进程内、进程间的实现方式。
同步阻塞是最经典的实现方式#xff0c;主要是为了…上一节课中我们学习了观察者模式的原理、实现、应用场景重点介绍了不同应用场景下几种不同的实现方式包括同步阻塞、异步非阻塞、进程内、进程间的实现方式。
同步阻塞是最经典的实现方式主要是为了代码解耦异步非阻塞除了能实现代码解耦之外还能提高代码的执行效率进程间的观察者模式解耦更加彻底一般是基于消息队列来实现用来实现不同进程间的被观察者和观察者之间的交互。
今天我们聚焦于异步非阻塞的观察者模式带你实现一个类似Google Guava EventBus的通用框架。等你学完本节课之后你会发现实现一个框架也并非一件难事。 对于上一节课的提到的采用异步非阻塞的形式处理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 UserService userService; // 依赖注入private ListRegObserver regObservers new ArrayList();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()函数中增加了这部分业务代码的维护成本。 EventBus框架功能需求介绍 EventBus翻译为“事件总线”它提供了实现观察者模式的骨架代码。我们可以基于此框架非常容易地在自己的业务场景中实现观察者模式不需要从零开始开发。其中Google Guava EventBus就是一个比较著名的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(ListObject observers) {for (Object observer : observers) {eventBus.register(observer);}}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.sendInboxMessage(userId, ...);}
}
利用EventBus框架实现的观察者模式跟从零开始编写的观察者模式相比从大的流程上来说实现思路大致一样都需要定义Observer并且通过register()函数注册Observer也都需要通过调用某个函数比如EventBus中的post()函数来给Observer发送消息在EventBus中消息被称作事件event。
但在实现细节方面它们又有些区别。基于EventBus我们不需要定义Observer接口任意类型的对象都可以注册到EventBus中通过Subscribe注解来标明类中哪个函数可以接收被观察者发送的消息。 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 Preconditions.checkNotNull(target);this.method method;this.method.setAccessible(true);}public void execute(Object event) { // event是method方法的参数try {method.invoke(target, event);} catch (InvocationTargetException | IllegalAccessException e) {e.printStackTrace();}}
}
3.ObserverRegistry
ObserverRegistry类就是前面讲到的Observer注册表是最复杂的一个类框架中几乎所有的核心逻辑都在这个类中。这个类大量使用了Java的反射语法不过代码整体来说都不难理解其中一个比较有技巧的地方是CopyOnWriteArraySet的使用。
public class ObserverRegistry {private ConcurrentMapClass?, 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 matchedObservers 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)) {matchedObservers.addAll(eventActions);}}return matchedObservers;}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 have exactly 1 parameter.,method, parameterTypes.length);annotatedMethods.add(method);}}return annotatedMethods;}
}
4.EventBus
EventBus实现的是阻塞同步的观察者模式。看代码你可能会有些疑问这明明就用到了线程池Executor啊。实际上MoreExecutors.directExecutor()是Google Guava提供的工具类看似是多线程实际上是单线程。之所以要这么实现主要还是为了跟AsyncEventBus统一代码逻辑做到代码复用。 public class EventBus {private Executor executor;private ObserverRegistry registry new ObserverRegistry();public EventBus() {this(MoreExecutors.directExecutor());}protected 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);}
}
