目前大多數開發者使用EventBus或者Otto作為事件總線通信庫，對于RxJava使用者來說，RxJava也可以輕松實現事件總線，因為它們都依據于觀察者模式。
不多說，上代碼
/*** RxBus* Created by YoKeyword on 2015/6/17.*/public class RxBus { private static volatile RxBus defaultInstance; private final Subject<Object, Object> bus; // PublishSubject只會把在訂閱發生的時間點之后來自原始Observable的數據發射給觀察者 public RxBus() { bus = new SerializedSubject<>(PublishSubject.create()); } // 單例RxBus public static RxBus getDefault() { if (defaultInstance == null) { synchronized (RxBus.class) { if (defaultInstance == null) { defaultInstance = new RxBus(); } } } return defaultInstance ; } // 發送一個新的事件 public void post (Object o) { bus.onNext(o); } // 根據傳遞的 eventType 類型返回特定類型(eventType)的 被觀察者 public <T> Observable<T> toObservable (Class<T> eventType) { return bus.ofType(eventType);// 這里感謝小鄧子的提醒: ofType = filter + cast// return bus.filter(new Func1<Object, Boolean>() {// @Override// public Boolean call(Object o) {// return eventType.isInstance(o);// }// }) .cast(eventType); }}
注：
1、Subject同時充當了Observer和Observable的角色，Subject是非線程安全的，要避免該問題，需要將 Subject轉換為一個 SerializedSubject ，上述RxBus類中把線程非安全的PublishSubject包裝成線程安全的Subject。
2、PublishSubject只會把在訂閱發生的時間點之后來自原始Observable的數據發射給觀察者。
3、ofType操作符只發射指定類型的數據，其內部就是filter+cast（這里非常感謝@小鄧子 的提醒）
public final <R> Observable<R> ofType(final Class<R> klass) { return filter(new Func1<T, Boolean>() { @Override public final Boolean call(T t) { return klass.isInstance(t); } }).cast(klass);}filter操作符可以使你提供一個指定的測試數據項，只有通過測試的數據才會被“發射”。
cast操作符可以將一個Observable轉換成指定類型的Observable。
分析：
RxBus工作流程圖
1、首先創建一個可同時充當Observer和Observable的Subject；
2、在需要接收事件的地方，訂閱該Subject（此時Subject是作為Observable），在這之后，一旦Subject接收到事件，立即發射給該訂閱者；
3、在我們需要發送事件的地方，將事件post至Subject，此時Subject作為Observer接收到事件（onNext），然后會發射給所有訂閱該Subject的訂閱者。
對于RxBus的使用，就和普通的RxJava訂閱事件很相似了。
先看發送事件的代碼：
RxBus.getDefault().post(new UserEvent (1, "yoyo"));
userEvent是要發送的事件，如果你用過EventBus, 很容易理解，UserEvent的代碼：
public class UserEvent { long id; String name; public UserEvent(long id,String name) { this.id= id; this.name= name; } public long getId() { return id; } public String getName() { return name; }}再看接收事件的代碼：
// rxSubscription是一個Subscription的全局變量，這段代碼可以在onCreate/onStart等生命周期內rxSubscription = RxBus.getDefault().toObserverable(UserEvent.class) .subscribe(new Action1<UserEvent>() { @Override public void call(UserEvent userEvent) { long id = userEvent.getId(); String name = userEvent.getName(); ... } }, new Action1<Throwable>() { @Override public void call(Throwable throwable) { // TODO: 處理異常 } });最后，一定要記得在生命周期結束的地方取消訂閱事件，防止RxJava可能會引起的內存泄漏問題。
@Overrideprotected void onDestroy() { super.onDestroy(); if(!rxSubscription.isUnsubscribed()) { rxSubscription.unsubscribe(); }}這樣，一個簡單的Event Bus就實現了！如果你的項目已經開始使用RxJava，也許可以考慮替換掉EventBus或Otto，減小項目體積。
RxBus、EventBus因為解耦太徹底，濫用的話，項目可維護性會越來越低；一些簡單場景更推薦用回調、Subject來代替事件總線。
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持VEVB武林網。
新聞熱點
疑難解答
