相信大家都對(duì)EventBus這個(gè)很熱門(mén)的工具很熟悉了，最近更新開(kāi)發(fā)新框架的緣故，接觸到Retrofit 2 和 RxJava 2 ，自然而然地使用RxBus來(lái)代替EventBus。

RxJava

一輛有意思的BUS

“Bus”這個(gè)詞其實(shí)很形象的說(shuō)明總線這個(gè)概念。那什么是總線？

總線是系統(tǒng)傳送信息的公共通信干線，各個(gè)部件在總線上傳遞消息。RxBus實(shí)現(xiàn)了一種事件總線，一條公共的通信通道，上面跑著各種信息，Android中的各個(gè)組件或是控件都可以向它發(fā)送各種信息，在各個(gè)組件或控件中只要訂閱這條總線，每當(dāng)總線收到消息的時(shí)候，這些訂閱者同樣就能收到這些消息。

RxBus 最基本的實(shí)現(xiàn)

支持注解方式訂閱事件。
支持粘滯事件。
支持3種Bus:
Publish Bus( RxBus，參見(jiàn) PublishSubject )
Behavior Bus(參見(jiàn) BehaviorSubject)
Replay Bus(參見(jiàn)ReplaySubject)

這三種都是基于Subject類(lèi)，它負(fù)責(zé)在非Rx-Apis之間充當(dāng)橋梁。但是也存在問(wèn)題。他沒(méi)有能力去處理onComplete 或 onError事件.如果在下游發(fā)生onError / onComplete ，上游將受到影響。

ReRelay開(kāi)源項(xiàng)目

JakeWharton 大神已經(jīng)幫你們處理好這些問(wèn)題。

Relays 是既是Observable也是Consumer的RxJava 類(lèi)型，它同樣能夠很容易在non-Rx api和 Rx api之間搭起橋梁,而不必要擔(dān)心下游的觸發(fā)的終止?fàn)顟B(tài)（onComplete 或 onError）。

RxBus 下載使用

Maven:

<dependency>
  <groupId>com.kw</groupId>
  <artifactId>rxbus</artifactId>
  <version>1.0.1</version>
  <type>pom</type>
</dependency>

Gradle:

compile 'com.kw:rxbus:1.0.1'

這個(gè)庫(kù) 并不含有 RxAndroid 和 RxJava ，請(qǐng)自行添加

如何使用

發(fā)送事件

RxBus.getInstance().send(new UserEvent(1,"名字"));

接受事件

//注冊(cè)
disposable = RxBus.getInstance().register(UserEvent.class, AndroidSchedulers.mainThread(), new Consumer<UserEvent>() {
            @Override
            public void accept(UserEvent userEvent) {
                btnNext.setText(userEvent.getName());
                Toast.makeText(getBaseContext(), userEvent.toString(), Toast.LENGTH_SHORT).show();
                Log.d("AActivity", "onNext:" + Thread.currentThread().getName());
                throw new NullPointerException("空指針錯(cuò)誤");//發(fā)生錯(cuò)誤之后，會(huì)取消訂閱
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                //發(fā)生錯(cuò)誤后僅僅會(huì)進(jìn)入一次，因?yàn)榘l(fā)生錯(cuò)誤之后，會(huì)取消訂閱
                Toast.makeText(getBaseContext(), throwable.getMessage(), Toast.LENGTH_SHORT).show();
            }
        });
        //上下兩段代碼具有相同意義
disposable=RxBus.getInstance().toObservable(UserEvent.class)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<UserEvent>() {
            @Override
            public void accept(UserEvent userEvent) throws Exception {
                btnNext.setText(userEvent.getName());
                Toast.makeText(getBaseContext(),userEvent.toString(),Toast.LENGTH_SHORT).show();
                Log.d("AActivity","onNext:"+Thread.currentThread().getName());
                throw new NullPointerException("空指針錯(cuò)誤");//發(fā)生錯(cuò)誤之后，會(huì)取消訂閱
            }
        },new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                //發(fā)生錯(cuò)誤后僅僅會(huì)進(jìn)入一次，因?yàn)榘l(fā)生錯(cuò)誤之后，會(huì)取消訂閱
                Toast.makeText(getBaseContext(),throwable.getMessage(),Toast.LENGTH_SHORT).show();
            }
        });

//解除
@Override
protected void onDestroy() {
    super.onDestroy();
    RxBus.getInstance().unregister(disposable);
}

關(guān)鍵代碼

public final class RxBus {
    private Relay<Object> bus = null;
    private static RxBus instance;

    //禁用構(gòu)造方法
    private RxBus() {
        bus = PublishRelay.create().toSerialized();
    }

    public static RxBus getInstance() {
        if (instance == null) {
            synchronized (RxBus.class) {
                if (instance == null) {
                    instance = new RxBus();
                }
            }
        }
        return instance;
    }

    public void send(Object event) {
        bus.accept(event);
    }

    public  <T> Observable<T> toObservable(Class<T> eventType) {
        return bus.ofType(eventType);
    }
    
    ...

}

使用時(shí)，需要你定義一些事件類(lèi)型，例如上面的UserEvent，該類(lèi)型只會(huì)傳遞給訂閱該類(lèi)型的觀察者。這主要?dú)w功于ofType操作符。

這里僅僅實(shí)現(xiàn)了廣播事件，基于ReRelay還可以實(shí)現(xiàn)其他方式。

如果有什么不理解的，請(qǐng)瀏覽基于Rxjava2+RxRelay 的RxBus源碼。