快捷導(dǎo)航

JAVA | Guava EventBus 使用發(fā)布/訂閱模式的步驟

更新時(shí)間：2021年03月02日 10:25:08 作者：雙鬼帶單

這篇文章主要介紹了JAVA | Guava EventBus 使用發(fā)布/訂閱模式的步驟，幫助大家更好的理解和學(xué)習(xí)使用Guava EventBus，感興趣的朋友可以了解下

前言

EventBus 是 Guava 的事件處理機(jī)制，是觀察者模式（生產(chǎn)/消費(fèi)模型）的一種實(shí)現(xiàn)。

觀察者模式在我們?nèi)粘ｉ_(kāi)發(fā)中使用非常廣泛，例如在訂單系統(tǒng)中，訂單狀態(tài)或者物流信息的變更會(huì)向用戶(hù)發(fā)送APP推送、短信、通知賣(mài)家、買(mǎi)家等等；審批系統(tǒng)中，審批單的流程流轉(zhuǎn)會(huì)通知發(fā)起審批用戶(hù)、審批的領(lǐng)導(dǎo)等等。

Observer模式也是 JDK 中自帶就支持的，其在 1.0 版本就已經(jīng)存在 Observer，不過(guò)隨著 Java 版本的飛速升級(jí)，其使用方式一直沒(méi)有變化，許多程序庫(kù)提供了更加簡(jiǎn)單的實(shí)現(xiàn)，例如 Guava EventBus、RxJava、EventBus 等

一、為什么要用 Observer模式以及 EventBus 優(yōu)點(diǎn) ？

EventBus 優(yōu)點(diǎn)

相比 Observer 編程簡(jiǎn)單方便
通過(guò)自定義參數(shù)可實(shí)現(xiàn)同步、異步操作以及異常處理
單進(jìn)程使用，無(wú)網(wǎng)絡(luò)影響

缺點(diǎn)

只能單進(jìn)程使用
項(xiàng)目異常重啟或者退出不保證消息持久化

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

二、EventBus 使用步驟

1. 引入庫(kù)

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>

引入依賴(lài)后，這里我們主要使用 com.google.common.eventbus.EventBus 類(lèi)進(jìn)行操作，其提供了 register、unregister、post 來(lái)進(jìn)行注冊(cè)訂閱、取消訂閱和發(fā)布消息

public void register(Object object);

public void unregister(Object object);

public void post(Object event);

2. 同步使用

1. 首先創(chuàng)建一個(gè) EventBus

EventBus eventBus = new EventBus();

2. 創(chuàng)建一個(gè)訂閱者

在 Guava EventBus 中，是根據(jù)參數(shù)類(lèi)型進(jìn)行訂閱，每個(gè)訂閱的方法只能由一個(gè)參數(shù)，同時(shí)需要使用 @Subscribe 標(biāo)識(shí)

class EventListener {

 /**
  * 監(jiān)聽(tīng) Integer 類(lèi)型的消息
  */
 @Subscribe
 public void listenInteger(Integer param) {
  System.out.println("EventListener#listenInteger ->" + param);
 }

 /**
  * 監(jiān)聽(tīng) String 類(lèi)型的消息
  */
 @Subscribe
 public void listenString(String param) {
  System.out.println("EventListener#listenString ->" + param);
 }
}

3. 注冊(cè)到 EventBus 上并發(fā)布消息

EventBus eventBus = new EventBus();

eventBus.register(new EventListener());

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

運(yùn)行結(jié)果為

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

根據(jù)需要我們可以創(chuàng)建多個(gè)訂閱者完成訂閱信息，同時(shí)如果一個(gè)類(lèi)型存在多個(gè)訂閱者，則所有訂閱方法都會(huì)執(zhí)行

為什么說(shuō)這么做是同步的呢？

Guava Event 實(shí)際上是使用線(xiàn)程池來(lái)處理訂閱消息的，通過(guò)源碼可以看出，當(dāng)我們使用默認(rèn)的構(gòu)造方法創(chuàng)建 EventBus 的時(shí)候，其中 executor 為 MoreExecutors.directExecutor()，其具體實(shí)現(xiàn)中直接調(diào)用的 Runnable#run 方法，使其仍然在同一個(gè)線(xiàn)程中執(zhí)行，所以默認(rèn)操作仍然是同步的，這種處理方法也有適用的地方，這樣既可以解耦也可以讓方法在同一個(gè)線(xiàn)程中執(zhí)行獲取同線(xiàn)程中的便利，比如事務(wù)的處理

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. 異步使用

通過(guò)上面的源碼，可以看出只要將構(gòu)造方法中的 executor 換成一個(gè)線(xiàn)程池實(shí)現(xiàn)即可, 同時(shí) Guava EventBus 為了簡(jiǎn)化操作，提供了一個(gè)簡(jiǎn)化的方案即 AsyncEventBus

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

這樣即可實(shí)現(xiàn)異步使用

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. 異常處理

如果處理時(shí)發(fā)生異常應(yīng)該如何處理? 在看源碼中，無(wú)論是 EventBus 還是 AsyncEventBus 都可傳入自定義的 SubscriberExceptionHandler 該 handler 當(dāng)出現(xiàn)異常時(shí)會(huì)被調(diào)用，我可可以從參數(shù) exception 獲取異常信息，從 context 中獲取消息信息進(jìn)行特定的處理

其接口聲明為

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

總結(jié)

在上面的基礎(chǔ)上，我們可以定義一些消息類(lèi)型來(lái)實(shí)現(xiàn)不同消息的監(jiān)聽(tīng)和處理，通過(guò)實(shí)現(xiàn) SubscriberExceptionHandler 來(lái)處理異常的情況，無(wú)論時(shí)同步還是異步都能游刃有余