JAVA | Guava EventBus 使用 發(fā)布/訂閱模式的步驟
前言
EventBus 是 Guava 的事件處理機(jī)制,是觀察者模式(生產(chǎn)/消費(fèi)模型)的一種實(shí)現(xiàn)。
觀察者模式在我們?nèi)粘i_發(fā)中使用非常廣泛,例如在訂單系統(tǒng)中,訂單狀態(tài)或者物流信息的變更會(huì)向用戶發(fā)送APP推送、短信、通知賣家、買家等等;審批系統(tǒng)中,審批單的流程流轉(zhuǎn)會(huì)通知發(fā)起審批用戶、審批的領(lǐng)導(dǎo)等等。
Observer模式也是 JDK 中自帶就支持的,其在 1.0 版本就已經(jīng)存在 Observer,不過隨著 Java 版本的飛速升級,其使用方式一直沒有變化,許多程序庫提供了更加簡單的實(shí)現(xiàn),例如 Guava EventBus、RxJava、EventBus 等
一、為什么要用 Observer模式以及 EventBus 優(yōu)點(diǎn) ?
EventBus 優(yōu)點(diǎn)
- 相比 Observer 編程簡單方便
- 通過自定義參數(shù)可實(shí)現(xiàn)同步、異步操作以及異常處理
- 單進(jìn)程使用,無網(wǎng)絡(luò)影響
缺點(diǎn)
- 只能單進(jìn)程使用
- 項(xiàng)目異常重啟或者退出不保證消息持久化
如果需要分布式使用還是需要使用 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
類進(jìn)行操作,其提供了 register
、unregister
、post
來進(jìn)行注冊訂閱、取消訂閱和發(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ù)類型進(jìn)行訂閱,每個(gè)訂閱的方法只能由一個(gè)參數(shù),同時(shí)需要使用 @Subscribe
標(biāo)識
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");
運(yùn)行結(jié)果為
EventListener#listenInteger ->1 EventListener#listenInteger ->2 EventListener#listenString ->3
根據(jù)需要我們可以創(chuàng)建多個(gè)訂閱者完成訂閱信息,同時(shí)如果一個(gè)類型存在多個(gè)訂閱者,則所有訂閱方法都會(huì)執(zhí)行
為什么說這么做是同步的呢?
Guava Event 實(shí)際上是使用線程池來處理訂閱消息的,通過源碼可以看出,當(dāng)我們使用默認(rèn)的構(gòu)造方法創(chuàng)建 EventBus
的時(shí)候,其中 executor
為 MoreExecutors.directExecutor()
,其具體實(shí)現(xiàn)中直接調(diào)用的 Runnable#run
方法,使其仍然在同一個(gè)線程中執(zhí)行,所以默認(rèn)操作仍然是同步的,這種處理方法也有適用的地方,這樣既可以解耦也可以讓方法在同一個(gè)線程中執(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 換成一個(gè)線程池實(shí)現(xiàn)即可, 同時(shí) Guava EventBus 為了簡化操作,提供了一個(gè)簡化的方案即 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)該如何處理? 在看源碼中,無論是 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ǔ)上,我們可以定義一些消息類型來實(shí)現(xiàn)不同消息的監(jiān)聽和處理,通過實(shí)現(xiàn) SubscriberExceptionHandler 來處理異常的情況,無論時(shí)同步還是異步都能游刃有余
參考
https://github.com/google/guava
https://github.com/greenrobot/EventBus
https://github.com/ReactiveX/RxJava
以上就是JAVA | Guava EventBus 使用 發(fā)布/訂閱模式的步驟的詳細(xì)內(nèi)容,更多關(guān)于Guava EventBus 使用 發(fā)布/訂閱模式的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
關(guān)于logback日志級別動(dòng)態(tài)切換的四種方式
這篇文章主要介紹了關(guān)于logback日志級別動(dòng)態(tài)切換的四種方式,具有很好的參考價(jià)值,希望對大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-08-08SpringSecurity使用PasswordEncoder加密用戶密碼的示例代碼
PasswordEncoder是Spring Security庫中的一個(gè)關(guān)鍵組件,它主要用于處理密碼的安全存儲(chǔ)和驗(yàn)證,本文將給大家介紹一下SpringSecurity使用PasswordEncoder加密用戶密碼的方法,需要的朋友可以參考下2024-09-09java文件如何統(tǒng)計(jì)字母出現(xiàn)的次數(shù)和百分比
這篇文章主要介紹了java文件如何統(tǒng)計(jì)字母出現(xiàn)的次數(shù)和百分比,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-11-11關(guān)于spring.factories失效原因分析及解決
這篇文章主要介紹了關(guān)于spring.factories失效原因分析及解決,具有很好的參考價(jià)值,希望對大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-07-07