Java | Guava EventBus 使用 发布/订阅模式



前言

EventBus 是 Guava 的事件处理机制,是观察者模式(生产/消费模型)的一种实现。

观察者模式在我们日常开发中使用非常广泛,例如在订单系统中,订单状态或者物流信息的变更会向用户发送APP推送、短信、通知卖家、买家等等;审批系统中,审批单的流程流转会通知发起审批用户、审批的领导等等。

Observer模式也是 JDK 中自带就支持的,其在 1.0 版本就已经存在 Observer,不过随着 Java 版本的飞速升级,其使用方式一直没有变化,许多程序库提供了更加简单的实现,例如 Guava EventBus、RxJava、EventBus 等

一、为什么要用 Observer模式以及 EventBus 优点 ?

EventBus 优点

  • 相比 Observer 编程简单方便
  • 通过自定义参数可实现同步、异步操作以及异常处理
  • 单进程使用,无网络影响

缺点

  • 只能单进程使用
  • 项目异常重启或者退出不保证消息持久化

如果需要分布式使用还是需要使用 MQ

二、EventBus 使用步骤

1. 引入库

Gradle

compile group: 'com.google.guava', name: 'guava', version: '29.0-jre'

Maven

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>29.0-jre</version>
</dependency>

引入依赖后,这里我们主要使用 com.google.common.eventbus.EventBus 类进行操作,其提供了 registerunregisterpost 来进行注册订阅、取消订阅和发布消息

public void register(Object object);

public void unregister(Object object);

public void post(Object event);

2. 同步使用

1. 首先创建一个 EventBus

EventBus eventBus = new EventBus();

2. 创建一个订阅者

在 Guava EventBus 中,是根据参数类型进行订阅,每个订阅的方法只能由一个参数,同时需要使用 @Subscribe 标识

class EventListener {

  /**
   * 监听 Integer 类型的消息
   */
  @Subscribe
  public void listenInteger(Integer param) {
    System.out.println("EventListener#listenInteger ->" + param);
  }

  /**
   * 监听 String 类型的消息
   */
  @Subscribe
  public void listenString(String param) {
    System.out.println("EventListener#listenString ->" + param);
  }
}

3. 注册到 EventBus 上并发布消息

EventBus eventBus = new EventBus();

eventBus.register(new EventListener());

eventBus.post(1);
eventBus.post(2);
eventBus.post("3");

运行结果为

EventListener#listenInteger ->1
EventListener#listenInteger ->2
EventListener#listenString ->3

根据需要我们可以创建多个订阅者完成订阅信息,同时如果一个类型存在多个订阅者,则所有订阅方法都会执行

为什么说这么做是同步的呢?

Guava Event 实际上是使用线程池来处理订阅消息的,通过源码可以看出,当我们使用默认的构造方法创建 EventBus 的时候,其中 executorMoreExecutors.directExecutor(),其具体实现中直接调用的 Runnable#run 方法,使其仍然在同一个线程中执行,所以默认操作仍然是同步的,这种处理方法也有适用的地方,这样既可以解耦也可以让方法在同一个线程中执行获取同线程中的便利,比如事务的处理

EventBus 部分源码

public class EventBus {
  private static final Logger logger = Logger.getLogger(EventBus.class.getName());
  private final String identifier;
  private final Executor executor;
  private final SubscriberExceptionHandler exceptionHandler;
  private final SubscriberRegistry subscribers;
  private final Dispatcher dispatcher;

  public EventBus() {
    this("default");
  }

  public EventBus(String identifier) {
    this(identifier, MoreExecutors.directExecutor(), Dispatcher.perThreadDispatchQueue(), EventBus.LoggingHandler.INSTANCE);
  }

  public EventBus(SubscriberExceptionHandler exceptionHandler) {
    this("default", MoreExecutors.directExecutor(), Dispatcher.perThreadDispatchQueue(), exceptionHandler);
  }

  EventBus(String identifier, Executor executor, Dispatcher dispatcher, SubscriberExceptionHandler exceptionHandler) {
    this.subscribers = new SubscriberRegistry(this);
    this.identifier = (String)Preconditions.checkNotNull(identifier);
    this.executor = (Executor)Preconditions.checkNotNull(executor);
    this.dispatcher = (Dispatcher)Preconditions.checkNotNull(dispatcher);
    this.exceptionHandler = (SubscriberExceptionHandler)Preconditions.checkNotNull(exceptionHandler);
  }
}

DirectExecutor 部分源码

enum DirectExecutor implements Executor {
  INSTANCE;

  private DirectExecutor() {
  }

  public void execute(Runnable command) {
    command.run();
  }

  public String toString() {
    return "MoreExecutors.directExecutor()";
  }
}

3. 异步使用

通过上面的源码,可以看出只要将构造方法中的 executor 换成一个线程池实现即可, 同时 Guava EventBus 为了简化操作,提供了一个简化的方案即 AsyncEventBus

EventBus eventBus = new AsyncEventBus(Executors.newCachedThreadPool());
  • 1

这样即可实现异步使用

AsyncEventBus 源码

public class AsyncEventBus extends EventBus {
  public AsyncEventBus(String identifier, Executor executor) {
    super(identifier, executor, Dispatcher.legacyAsync(), LoggingHandler.INSTANCE);
  }

  public AsyncEventBus(Executor executor, SubscriberExceptionHandler subscriberExceptionHandler) {
    super("default", executor, Dispatcher.legacyAsync(), subscriberExceptionHandler);
  }

  public AsyncEventBus(Executor executor) {
    super("default", executor, Dispatcher.legacyAsync(), LoggingHandler.INSTANCE);
  }
}

4. 异常处理

如果处理时发生异常应该如何处理? 在看源码中,无论是 EventBus 还是 AsyncEventBus 都可传入自定义的 SubscriberExceptionHandler 该 handler 当出现异常时会被调用,我可可以从参数 exception 获取异常信息,从 context 中获取消息信息进行特定的处理

其接口声明为

public interface SubscriberExceptionHandler {
  /** Handles exceptions thrown by subscribers. */
  void handleException(Throwable exception, SubscriberExceptionContext context);
}

总结

在上面的基础上,我们可以定义一些消息类型来实现不同消息的监听和处理,通过实现 SubscriberExceptionHandler 来处理异常的情况,无论时同步还是异步都能游刃有余

参考

  • https://github.com/google/guava
  • https://github.com/greenrobot/EventBus
  • https://github.com/ReactiveX/RxJava

帖子状态

话题参与者

回复显示排序:

给大家安利一款免费的在线流程图设计软件:贼好用 ProcessFlow — 免费在线作图、实时协作 ProcessFlow是一个在线作图工具的聚合平台, 它可以在线画流程图、思维导图、UI原型图、UML、网络拓扑图、组织结构图等等, 您无需担心下载和更新的问题, 不管Mac还是Windows,一个浏览器就可以随时随地的发挥创意,规划工作,解放您的双手,让您腾出双手去成就别人的梦想。

Finally!
有没有特别的设计建议或更新的指南?
Finally!
您好,这边没有的。
tesla02 best answer
6 Jan,2019
Finally!
有没有特别的设计建议或更新的指南?
此帖子已被版主标记,收到太多否决票。

这是太大的预览图像,它应该更小,甚至五行。在一个页面上有30到60个页面,它是1800个项目 类别例如在附加组件类别有22749个项目,为什么不看到所有的项目,但只有那1800个项目?这是件坏事。

看起来您是个新人,欢迎免费注册社区,加入我们的大家庭,一起学习,一起成长。
提交你的评论
System.out.println("请输入你的代码块...");
话题
分类
喜欢
回复
浏览
活跃
原创社区
985
0
1.5k
原创社区
985
1
1.5k
20/12/25 22:32:21
javaWeb
985
1
1.5k
20/06/15 22:09:21
youtube
698
78
2.1k
3d
Settings

请插入代码块