前言

由於最近在學RxJava2，同樣避免不了使用Rxbus,但是RxJava1到RxJava2還是有很大的變化的，例如：在 2.x 中 Observable 不支援背壓了，將用一個全新的 Flowable 來支援背壓，Consumer和BiConsumer對Action1 和Action2進行了替換，Function 替換了Func1，採用BiFunction 替換Func 2..N。並且，它們都增加了throws Exception，也就是說，媽媽再也不用擔心我們做某些操作還需要try-catch了。所以我們以前的基於Rxjava1的RxBus就要升級改造下。

RxBus版本一

對於RxJava1.x.x，相信大家都已經用的很熟了，RxBus也不在話下，下面是本人在RxJava1版本使用的RxJava，程式碼裡有很詳細的註釋。這裡就不多說了！

• RxBus1

public class RxBus {

//    private static volatile RxBus mInstance;
    private SerializedSubject<Object, Object> mSubject;
    private HashMap<String, CompositeSubscription> mSubscriptionMap;

    /**
     *
     * 1、Subject同時充當了Observer和Observable的角色，Subject是非執行緒安全的，要避免該問題，
     * 需要將 Subject轉換為一個 SerializedSubject，類中把執行緒非安全的PublishSubject包裝成執行緒安全的Subject。
     * 2、PublishSubject只會把在訂閱發生的時間點之後來自原始Observable的資料發射給觀察者。
     *
     */
 

    public RxBus() {
        mSubject = new SerializedSubject<>(PublishSubject.create());
    }

    // 這裡使用Dagger來處理單例
    /*
    private RxBus() {
        mSubject = new SerializedSubject<>(PublishSubject.create());
    }

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


    /**
     * 傳送事件
     *
     * @param o
     */
    public void post(Object o) {
        mSubject.onNext(o);
    }

    /**
     * 返回指定型別的Observable例項
     * 1、ofType操作符只發射指定型別的資料，其內部就是filter+cast
     * @param type
     * @param <T>
     * @return
     */
    public <T> Observable<T> toObservable(final Class<T> type) {
        return mSubject.ofType(type);
    }

    /**
     * 是否已有觀察者訂閱
     *
     * @return
     */
    public boolean hasObservers() {
        return mSubject.hasObservers();
    }

    /**
     * 一個預設的訂閱方法
     *
     * @param type
     * @param next
     * @param error
     * @param <T>
     * @return
     */
    public <T> Subscription doSubscribe(Class<T> type, Action1<T> next, Action1<Throwable> error) {
        return toObservable(type)
                // 加上背壓處理，不然有些地方會有異常，關於背壓參考這裡：https://gold.xitu.io/post/582d413c8ac24700619cceed
                .onBackpressureBuffer()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(next, error);

    }

   /**
     * 儲存訂閱後的subscription
     * @param o
     * @param subscription
     */
    public void addSubscription(Object o, Subscription subscription) {
        if (mSubscriptionMap == null) {
            mSubscriptionMap = new HashMap<>();
        }
        String key = o.getClass().getName();
        if (mSubscriptionMap.get(key) != null) {
            mSubscriptionMap.get(key).add(subscription);
        } else {
//            代表一組訂閱,訂閱 unsubscribed在一起
            CompositeSubscription compositeSubscription = new CompositeSubscription();
            compositeSubscription.add(subscription);
            mSubscriptionMap.put(key, compositeSubscription);
        }
    }
    /**
     * 取消訂閱
     * @param o
     */
    public void unSubscribe(Object o) {
        if (mSubscriptionMap == null) {
            return;
        }
        String key = o.getClass().getName();
        if (!mSubscriptionMap.containsKey(key)){
            return;
        }
        if (mSubscriptionMap.get(key) != null) {
            mSubscriptionMap.get(key).unsubscribe();
        }
        mSubscriptionMap.remove(key);
    }
}

RxBus版本二

在這之前我們先熟悉下EventBus和RxJava的原理，這裡只是簡單聊聊：
EventBus
EventBus是Android下高效的釋出/訂閱事件匯流排機制,是一種釋出訂閱設計模式（Publish/Subsribe），或稱作觀察者設計模式。
RxJava
RxJava 的非同步實現，是通過一種擴充套件的觀察者模式來實現的。RxJava 有四個基本概念：Observable (可觀察者，即被觀察者)、 Observer (觀察者)、 subscribe (訂閱)、事件。Observable 和 Observer 通過 subscribe() 方法實現訂閱關係，從而 Observable 可以在需要的時候發出事件來通知 Observer。
明白了什麼是觀察者模式，我們才能更好地對RxBbus1改造成我們需要的RxBus2.

• RxBus2
  翻Rxjava2的文件，我們知道在RxJava2中已經沒了SerializedSubject類，我們可以呼叫toSerialized()方法，對Subject包裝序列化，使其變成執行緒安全的。由於Rxjava2使用Flowable支援背壓，我們可以將其轉換成帶背壓的Flowable例項。基本上bus1和bus2的區別就這些！

public class RxBus {
    private HashMap<String, CompositeDisposable> mSubscriptionMap;
    private static volatile RxBus mRxBus;
    private final Subject<Object> mSubject;
    //單列模式
    public static RxBus getIntanceBus(){
        if (mRxBus==null){
            synchronized (RxBus.class){
                if(mRxBus==null){
                    mRxBus = new RxBus();
                }
            }
        }
        return mRxBus;
    }
    public RxBus(){
         mSubject = PublishSubject.create().toSerialized();
    }

    public void post(Object o){
        mSubject.onNext(o);
    }
    /**
     * 返回指定型別的帶背壓的Flowable例項
     *
     * @param <T>
     * @param type
     * @return
     */
    public <T>Flowable<T> getObservable(Class<T> type){
        return mSubject.toFlowable(BackpressureStrategy.BUFFER)
                .ofType(type);
    }
    /**
     * 一個預設的訂閱方法
     *
     * @param <T>
     * @param type
     * @param next
     * @param error
     * @return
     */
    public <T> Disposable doSubscribe(Class<T> type, Consumer<T> next, Consumer<Throwable> error){
        return getObservable(type)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(next,error);
    }
    /**
     * 是否已有觀察者訂閱
     *
     * @return
     */
    public boolean hasObservers() {
        return mSubject.hasObservers();
    }
    /**
     * 儲存訂閱後的disposable
     * @param o
     * @param disposable
     */
    public void addSubscription(Object o, Disposable disposable) {
        if (mSubscriptionMap == null) {
            mSubscriptionMap = new HashMap<>();
        }
        String key = o.getClass().getName();
        if (mSubscriptionMap.get(key) != null) {
            mSubscriptionMap.get(key).add(disposable);
        } else {
            //一次性容器,可以持有多個並提供 新增和移除。
            CompositeDisposable disposables = new CompositeDisposable();
            disposables.add(disposable);
            mSubscriptionMap.put(key, disposables);
        }
    }

    /**
     * 取消訂閱
     * @param o
     */
    public void unSubscribe(Object o) {
        if (mSubscriptionMap == null) {
            return;
        }

        String key = o.getClass().getName();
        if (!mSubscriptionMap.containsKey(key)){
            return;
        }
        if (mSubscriptionMap.get(key) != null) {
            mSubscriptionMap.get(key).dispose();
        }

        mSubscriptionMap.remove(key);
    }
}

• 使用

傳送事件

RxBus rxBus = RxBus.getIntanceBus();
//傳送事件
rxBus.post(new RxBusMessage("1"));
rxBus.post(new RxBusMessage("2"));

接受事件

private void initRxBus() {
        rxBus = RxBus.getIntanceBus();
        registerRxBus(RxBusMessage.class, new Consumer<RxBusMessage>() {
            @Override
            public void accept(@NonNull RxBusMessage rxBusMessage) throws Exception {
             //根據事件型別進行處理
                if(TextUtils.equals(rxBusMessage.getMessage(),"1")){
                    mTvRebus.setText(rxBusMessage.getMessage());
                }else if(TextUtils.equals(rxBusMessage.getMessage(),"2")){
                    rx2.setText(rxBusMessage.getMessage());
                }
            }
        });
    }

//註冊
    public <T> void registerRxBus(Class<T> eventType, Consumer<T> action) {
        Disposable disposable = rxBus.doSubscribe(eventType, action, new Consumer<Throwable>() {
            @Override
            public void accept(@NonNull Throwable throwable) throws Exception {
                Log.e("NewsMainPresenter", throwable.toString());
            }
        });
        rxBus.addSubscription(this,disposable);
    }

銷燬事件

 public void unregisterRxBus() {
        rxBus.unSubscribe(this);
    }
@Override
    protected void onDestroy() {
        super.onDestroy();
        unregisterRxBus();
    }

效果

最後

有什麼不對的地方敬請指正！大家一起進步撒！

拼搏在技術道路上的一隻小白