用RxJava實現(xiàn)事件總線(Event Bus)

目前大多數(shù)開發(fā)者使用EventBus或者Otto作為事件總線通信庫,對于RxJava使用者來說,RxJava也可以輕松實現(xiàn)事件總線,因為它們都依據(jù)于觀察者模式。

不多說,上代碼

/**
* RxBus
* Created by YoKeyword on 2015/6/17.
*/
public class RxBus {
    private static volatile RxBus defaultInstance;

    private final Subject<Object, Object> bus;
    // PublishSubject只會把在訂閱發(fā)生的時間點之后來自原始Observable的數(shù)據(jù)發(fā)射給觀察者
    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 ;
    }
    // 發(fā)送一個新的事件
    public void post (Object o) {
        bus.onNext(o);
    }
    // 根據(jù)傳遞的 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轉(zhuǎn)換為一個 SerializedSubject
,上述RxBus類中把線程非安全的PublishSubject包裝成線程安全的Subject。

2、PublishSubject只會把在訂閱發(fā)生的時間點之后來自原始Observable的數(shù)據(jù)發(fā)射給觀察者。

3、ofType操作符只發(fā)射指定類型的數(shù)據(jù),其內(nèi)部就是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操作符可以使你提供一個指定的測試數(shù)據(jù)項,只有通過測試的數(shù)據(jù)才會被“發(fā)射”。
cast操作符可以將一個Observable轉(zhuǎn)換成指定類型的Observable。

分析:

RxBus工作流程圖

1、首先創(chuàng)建一個可同時充當Observer和Observable的Subject;

2、在需要接收事件的地方,訂閱該Subject(此時Subject是作為Observable),在這之后,一旦Subject接收到事件,立即發(fā)射給該訂閱者;

3、在我們需要發(fā)送事件的地方,將事件post至Subject,此時Subject作為Observer接收到事件(onNext),然后會發(fā)射給所有訂閱該Subject的訂閱者。

對于RxBus的使用,就和普通的RxJava訂閱事件很相似了。
先看發(fā)送事件的代碼:

RxBus.getDefault().post(new UserEvent (1, "yoyo"));

userEvent是要發(fā)送的事件,如果你用過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等生命周期內(nèi)
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: 處理異常
            }        
        });

最后,一定要記得在生命周期結(jié)束的地方取消訂閱事件,防止RxJava可能會引起的內(nèi)存泄漏問題。

@Override
protected void onDestroy() {
    super.onDestroy();
    if(!rxSubscription.isUnsubscribed()) {
        rxSubscription.unsubscribe();
    }
}

這樣,一個簡單的Event Bus就實現(xiàn)了!如果你的項目已經(jīng)開始使用RxJava,也許可以考慮替換掉EventBus或Otto,減小項目體積。

RxBus、EventBus因為解耦太徹底,濫用的話,項目可維護性會越來越低;一些簡單場景更推薦用回調(diào)、Subject來代替事件總線。

感興趣的可以閱讀我另外2篇深入RxBus的文章:
[深入RxBus:[支持Sticky事件]](http://m.itdecent.cn/p/71ab00a2677b)
深入RxBus:[異常處理]

參考:
http://nerds.weddingpartyapp.com/tech/2014/12/24/implementing-an-event-bus-with-rxjava-rxbus/

最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點,簡書系信息發(fā)布平臺,僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容