如果你的專案中用到了Rxjava，那麼你就完全可以自己去寫一個RxBus來替換EventBus，讓專案的體積變小。

在網上看了很多人寫的RxBus，總感覺有些缺陷或者不能完全替換EventBus的功能。所以，綜合別人寫的RxBus我自己實現了一個比較完整的RxBus。

首先說說EventBus的功能：
1，註冊、反註冊、傳送資料
2，接收資料（根據型別接收）
3，不同執行緒接收資料（非ui→非ui、非ui→ui、ui→非ui）
4，觀察者只關心傳送資料，不關心有多少訂閱者

那麼我們使用Rxjava也需要來完成這幾個功能才算真正的替換。

我們先定義一個RxBus類，並寫成單例。

/**
 * rxbus的核心類
 */
 

public class RxBus {

    public static RxBus getDefault() {
        return RxBusInstance.rxBus;
    }
    //內部類
    private static class RxBusInstance {
        private static final RxBus rxBus = new RxBus();
    }
    // 主題，Subject是非執行緒安全的
    public final Subject bus;

    /**
     * 單例
     * PublishSubject只會把在訂閱發生的時間點之後來自原始Observable的資料發射給觀察者
     * 序列化主題
     * 將 Subject轉換為一個 SerializedSubject ，類中把執行緒非安全的PublishSubject包裝成執行緒安全的Subject
     */
 

    private RxBus() {
        bus = new SerializedSubject<>(PublishSubject.create());
    }
}

首相我們得到一個主題（SerializedSubject），這個主題主要的作用就是傳送資料。

其中，觀察者 Observable是在使用者註冊的時候得到的物件，它用於接收觀察者發出的資料。
訂閱物件Subscription用於反註冊和判斷該物件是否已經被訂閱

接下來，我們先寫
註冊：

    /// 根據傳遞的 eventType 型別返回特定型別(eventType)的 被觀察者
    public
 
 Observable register(Class eventType) {
        return bus.ofType(eventType);
    }

傳送資料：

    // 提供了一個新的事件用於傳送,這時候的Subject是一個觀察者
    //ofType將返回一個特定的class型別
    public void post(final Object obj) {
        bus.onNext(obj);
    }

這就是所有的方法，很簡單，只有註冊和傳送資料。
另外還有一些方法例如接收方法，我們可以根據Rxjava的執行緒規劃來寫出其他的方式。
但不是在這個類中完成。

接下來我們來看一看它的用法：

我們可以在網路執行緒等耗時操作中傳送資料：

                //傳送訊息
                RxBus.getDefault().post(UserInfoBean);

然後在activity或者fragment的onResume方法中註冊

@Override
    protected void onResume() {
      Observable observable=  RxBus.getDefault().register(UserInfoBean.class);//註冊使用者資訊事件
        super.onResume();
    }

在activity或者fragment的onStop方法中反註冊

@Override
    protected void onStop() {
        if (!observable.isUnsubscribed())
            observable.unregister();//如果訂閱取消訂閱
        super.onStop();
    }

然後不管在activity的哪個地方都能夠去獲取資料：

//接收訊息，我們不需要關心觀察者，只要我們通過observable物件就能得到RxBus傳送的資料
        Subscription subscriptionUser = observable
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<UserInfoBean>() {
                    @Override
                    public void call(UserInfoBeans) {
                        main_text.append(s.getUserName());
                    }
                });//接收String型別的資料

或者在其他執行緒

.observeOn(Schedulers.newThread())
.observeOn(Schedulers.io())
//等等

寫到這裡我想各位看客應該能夠理解這個RxBus是如何實現了的吧？

我們來總結一下：

如果我訂閱了UserInfoBean的資料型別的話，那麼RxBus傳送userInfoBean的資料的時候，我就能獲取到。不管RxBus寫在哪裡，不管我具體在哪裡需要。
我們只需要從在observable物件中傳入UserInfoBean，就可以得到它的資料了。

1，我們在初始化的時候例項化一個主題：

bus = new SerializedSubject<>(PublishSubject.create());

這個主題必須是執行緒安全的。也就是說它是SerializedSubject而不是Subject。

2，這個主題用於向所有訂閱者傳送資料：

bus.onNext(obj);

3，註冊的時候過濾傳送型別。我們傳入一個class物件，讓這個主題知道，我將要傳送這個型別的資料，其他的不會發送。

 bus.ofType(cla);

4，於此同時，我們在具體使用到的地方例項化（註冊）observable這個觀察者例項用於例項化一個訂閱物件和告訴訂閱者資料來了！

Observable observable = RxBus.getDefault()
                .register(String.class);

5，我們從observable物件中獲得的訂閱物件Subscription，用於獲取觀察者傳送過來的資料

Subscription subscription = observable
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        main_text.append(s);
                    }
                });//接收String型別的資料

5，獲得的訂閱物件用於判斷這個訂閱物件是否被訂閱以及取消訂閱。

 //反註冊，將註冊的subscription物件退訂
 //是否沒訂閱
        Toast.makeText(this, !subscription.isUnsubscribed() ? "已經訂閱" : "未訂閱", Toast.LENGTH_LONG).show();
        //取消訂閱
        subscription.unsubscribe();

這個RxBus類就寫完了，我們也可用它做EventBus的事情了。
不過……雖然這個類是我在看了他人的RxBus類後改進和改良的。

但是它依然存在問題。

可以看出來，我們做了很多重複的操作。如果我們在一個activity裡面需要接受UserInfoBean、IndexInfoBean等等，那麼我們就要有幾個資料接收就要管理幾個觀察者和訂閱者。

這對於一個優秀的程式碼來說顯然是不夠美好的。

不過，為什麼不能將他們封裝呢？我試過很多方法都無法封裝。
因為，我們要依靠它的泛型例項化來確定這個觀察者傳送的是什麼型別的資料。
如果我們封裝了觀察者，那麼很多問題很難解決，至少目前為止我沒有解決。

1，如何將具體型別傳進去替換掉泛型？
2，一個數據結構對應一個觀察者。觀察者如何分身多個？
3，一個觀察者對應了多個訂閱者，如何統一管理和區分登出訂閱者？

至少目前為止，只能這麼寫。如果有哪位看客有更好的封裝方法，歡迎探討！