观察者模式(下):如何实现一个异步非阻塞的EventBus框架?

这篇具有很好参考价值的文章主要介绍了观察者模式(下):如何实现一个异步非阻塞的EventBus框架?。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

上一节课中,我们学习了观察者模式的原理、实现、应用场景,重点介绍了不同应用场景下,几种不同的实现方式,包括:同步阻塞、异步非阻塞、进程内、进程间的实现方式。

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

今天,我们聚焦于异步非阻塞的观察者模式,带你实现一个类似Google Guava EventBus的通用框架。等你学完本节课之后,你会发现,实现一个框架也并非一件难事。

对于上一节课的提到的采用异步非阻塞的形式处理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 UserService userService; // 依赖注入
  private List<RegObserver> regObservers = new ArrayList<>();
  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()函数中,增加了这部分业务代码的维护成本。

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(List<Object> 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; // 依赖注入

  @Subscribe
  public void handleRegSuccess(Long userId) {
    promotionService.issueNewUserExperienceCash(userId);
  }
}

public class RegNotificationObserver {
  private NotificationService notificationService;

  @Subscribe
  public void handleRegSuccess(Long userId) {
    notificationService.sendInboxMessage(userId, "...");
  }
}

利用EventBus框架实现的观察者模式,跟从零开始编写的观察者模式相比,从大的流程上来说,实现思路大致一样,都需要定义Observer,并且通过register()函数注册Observer,也都需要通过调用某个函数(比如,EventBus中的post()函数)来给Observer发送消息(在EventBus中消息被称作事件event)。

但在实现细节方面,它们又有些区别。基于EventBus,我们不需要定义Observer接口,任意类型的对象都可以注册到EventBus中,通过@Subscribe注解来标明类中哪个函数可以接收被观察者发送的消息。

观察者模式(下):如何实现一个异步非阻塞的EventBus框架?,设计模式,观察者模式,java,服务器

观察者模式(下):如何实现一个异步非阻塞的EventBus框架?,设计模式,观察者模式,java,服务器

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 ConcurrentMap<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> matchedObservers = 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)) {
        matchedObservers.addAll(eventActions);
      }
    }
    return matchedObservers;
  }

  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 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) {
    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()了,而是需要在构造函数中,由调用者注入线程池。文章来源地址https://www.toymoban.com/news/detail-571161.html

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

到了这里,关于观察者模式(下):如何实现一个异步非阻塞的EventBus框架?的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Go和Java实现观察者模式

    在监控系统中,我们需要采集监控指标的信息,假设当采集的指标信息超过阈值时我们需要对该监控指标持久化到 数据库中并且进行告警。 本文通过指标采集持久化和告警来说明观察者模式的使用,使用Go语言和Java语言实现。 观察者模式是一种行为型设计模式,它定义了一

    2024年02月16日
    浏览(35)
  • 如何用C++11实现观察者模式

    观察者模式是一种设计模式,定义了对象之间的一对多关系。当一个对象状态发生改变时,它的依赖者都会收到通知并自动更新。在C++11中,可以通过以下方式实现观察者模式: 首先,我们需要创建一个观察者接口,其中包含一个更新方法。这个接口可以被多个观察者类实现

    2024年02月15日
    浏览(32)
  • 「观察者(Observer)」设计模式 Swift实现

    观察者设计模式(Observer Pattern)是一种行为型设计模式,它定义了对象之间的一种一对多的依赖关系,使得当一个对象的状态发生变化时,所有依赖于它的对象都会受到通知并自动更新。 在这种模式中,被观察者对象(Subject)存储其观察者对象(Observer)列表,并提供了用

    2024年02月16日
    浏览(39)
  • ts:简单实现 mobx 的观察者模式 observable

    mobx 状态管理器 订阅者: 管理 变化后 回调的函数列表 观察者:将传入的 数据监听 起来,并且在 有变化时 , 执行 订阅者管理的 函数列表

    2024年02月12日
    浏览(53)
  • 【设计模式】Head First 设计模式——观察者模式 C++实现

    设计模式最大的作用就是在变化和稳定中间寻找隔离点,然后分离它们,从而管理变化。将变化像小兔子一样关到笼子里,让它在笼子里随便跳,而不至于跳出来把你整个房间给污染掉。 主题对象(出版者)管理某些数据,当主题内的数据改变,就会通知观察者(订阅者)。

    2024年02月10日
    浏览(41)
  • 性能优化之懒加载 - 基于观察者模式和单例模式的实现

            在前端性能优化中,关于图片/视频等内容的懒加载一直都是优化利器。当用户看到对应的视图模块时,才去请求加载对应的图像。 原理也很简单,通过浏览器提供的 IntersectionObserver - Web API 接口参考 | MDN (mozilla.org),观察“哪个元素和视口交叉”,从而进行懒加载。

    2024年02月05日
    浏览(37)
  • 用C++11打造智能观察者模式:详解实现步骤完整示例代码

      观察者模式是一种行为设计模式,其中一个对象(主题)维护其依赖对象(观察者)的列表,当主题的状态发生变化时,它通知所有观察者。以下是一个使用C++11实现观察者模式的简单例子: 定义观察者接口(Observer):  创建一个观察者接口,该接口包含观察者需要实现

    2024年01月25日
    浏览(48)
  • 【C++ 观察者模式 思想理解】C++中的观察者模式:松耦合设计与动态交互的艺术,合理使用智能指针观察者

    在进入技术细节之前,理解观察者模式(Observer Pattern)的基本概念和它在现代编程中的重要性是至关重要的。 观察者模式是一种设计模式,它定义了对象间的一种一对多的依赖关系,当一个对象的状态发生改变时,所有依赖于它的对象都得到通知并自动更新。在C++中,这个

    2024年01月24日
    浏览(56)
  • 行为型模式 | 观察者模式

    观察者模式又叫做发布-订阅(Publish/Subscribe)模式,定义了一种一对多的依赖关系 。让多个观察者对象同时监听某一个主题对象,这个主题对象在状态上发生变化时,会通知所有观察者对象,使它们能够自动更新自己。 软件系统常常要求在某一个对象的状态发生变化的时候

    2024年01月22日
    浏览(79)
  • 【设计模式】观察者模式

    观察者模式(又被称为发布-订阅(Publish/Subscribe)模式,属于行为型模式的一种,它定义了一种一对多的依赖关系,让多个观察者对象同时监听某一个主题对象。这个主题对象在状态变化时,会通知所有的观察者对象,使他们能够自动更新自己。 Subject:抽象主题(被观察者

    2024年02月13日
    浏览(52)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包