前言

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

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

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

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

EventBus 優(yōu)點

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

缺點

只能單進程使用 項目異常重啟或者退出不保證消息持久化

如果需要分布式使用還是需要使用 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 類進行操作，其提供了 register、unregister、post 來進行注冊訂閱、取消訂閱和發(fā)布消息

public void register(Object object);public void unregister(Object object);public void post(Object event);2. 同步使用

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

EventBus eventBus = new EventBus();

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

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

class EventListener { /** * 監(jiān)聽 Integer 類型的消息 */ @Subscribe public void listenInteger(Integer param) { System.out.println('EventListener#listenInteger ->' + param); } /** * 監(jiān)聽 String 類型的消息 */ @Subscribe public void listenString(String param) { System.out.println('EventListener#listenString ->' + param); }}

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

EventBus eventBus = new EventBus();eventBus.register(new EventListener());eventBus.post(1);eventBus.post(2);eventBus.post('3');

運行結(jié)果為

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

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

為什么說這么做是同步的呢？

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

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

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

這樣即可實現(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. 異常處理

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

其接口聲明為

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

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

參考

https://github.com/google/guavahttps://github.com/greenrobot/EventBushttps://github.com/ReactiveX/RxJava

以上就是JAVA | Guava EventBus 使用 發(fā)布/訂閱模式的步驟的詳細內(nèi)容，更多關(guān)于Guava EventBus 使用 發(fā)布/訂閱模式的資料請關(guān)注好吧啦網(wǎng)其它相關(guān)文章！