RxBus在rxjava:2.*** 的基本使用

原文地址：http://blog.csdn.net/donkor_/article/details/79709366

▲ 前言：

事件总线的好处在于方便组件之间的交互，RxBus不是一个库，而是使用RxJava实现事件总线的一种思想。刚好项目中使用到了rxjava，相对于使用eventbus，使用rxbus来的更方便。

▲ rxbus和eventbus相比较：

RxJava 主要做异步、网络的数据处理，强大之处就是对数据的处理了，而对于处理完后的数据处理是一样的都是观察者模式来通知，也可以把 RxJava 进一步封装出一个 EventBus（RxBus）库，二者可以转换的。

EventBus比较适合仅仅当做组件间的通讯工具使用，主要用来传递消息。使用EventBus可以避免搞出一大推的interface，仅仅是为了实现组件间的通讯，而不得不去实现那一推的接口

▲ rxbus实现原理：

Note that it is important to subscribe to the exact same rxBus instance that was used to post the events

▲ 基本配置：

//引入rxJava
implementation 'io.reactivex.rxjava2:rxjava:2.2.3'
//引入rxAndroid
implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'
//引入J神的rxrelay2，出现异常仍然可以处理
implementation 'com.jakewharton.rxrelay2:rxrelay:2.0.0'

▲ 基本使用一：

在订阅者处理事件出现异常后，订阅者无法再收到事件，这是 RxJava 当初本身的设计原则，但是在事件总线中这反而是个问题，不过 JakeWharton 大神写了即使出现异常也不会终止订阅关系的 RxRelay，所以基于 RxRelay 就能写出有异常处理能力的 Rxbus。

/**

有异常处理的 Rxbus
@author Donkor
*/

public class RxBus {
private static volatile RxBus instance;
private final Relay<Object> mBus;

public RxBus() {
    this.mBus = PublishRelay.create().toSerialized();
}

public static RxBus getInstance() {
    if (instance == null) {
        synchronized (RxBus.class) {
            if (instance == null) {
                instance = Holder.BUS;
            }
        }
    }
    return instance;
}
public void post(Object obj) {
    mBus.accept(obj);
}

public <T> Observable<T> toObservable(Class<T> tClass) {
    return  mBus.ofType(tClass);
}

public Observable<Object> toObservable() {
    return mBus;
}

public boolean hasObservers() {
    return mBus.hasObservers();
}

private static class Holder {
    private static final RxBus BUS = new RxBus();
}

}

关于rxbus异常处理还不太了解的同学可以参考一下这篇文章：[深入RxBus]：异常处理
https://www.jianshu.com/p/0493cc28a811

▲ 基本使用二：

没有背压处理（Backpressure）的 Rxbus

/**

@author Donkor
*/
public class RxBus {

private final Subject<Object> mBus;

private RxBus() {
// toSerialized method made bus thread safe
mBus = PublishSubject.create().toSerialized();
}

public static RxBus get() {
return Holder.BUS;
}

public void post(Object obj) {
mBus.onNext(obj);
}

public <T> Observable<T> toObservable(Class<T> tClass) {
return mBus.ofType(tClass);
}

public Observable<Object> toObservable() {
return mBus;
}

public boolean hasObservers() {
return mBus.hasObservers();
}

private static class Holder {
private static final RxBus BUS = new RxBus();
}
}

有背压处理（Backpressure）的 Rxbus

/**

@author Donkor
*/
public class RxBus {

private final FlowableProcessor<Object> mBus;

private RxBus() {
// toSerialized method made bus thread safe
mBus = PublishProcessor.create().toSerialized();
}

public static RxBus get() {
return Holder.BUS;
}

public void post(Object obj) {
mBus.onNext(obj);
}

public <T> Flowable<T> toFlowable(Class<T> tClass) {
return mBus.ofType(tClass);
}

public Flowable<Object> toFlowable() {
return mBus;
}

public boolean hasSubscribers() {
return mBus.hasSubscribers();
}

private static class Holder {
private static final RxBus BUS = new RxBus();
}
}

关于有背压与没有背压的知识介绍，可以参考下这一篇文章关于RxJava背压
http://flyou.ren/2017/04/05/%E5%85%B3%E4%BA%8ERxJava%E8%83%8C%E5%8E%8B/?utm_source=tuicool&utm_medium=referral

▲ 举个栗子：

场景：当前有三个（多个）activity，打开第二、三个（多个）activity需要修改第一个activity的数据时，这时就可以使用rxbus，相比发送广播速度快。

无图无真相，直接看下效果图

第一个页面关键代码

RxBus.getInstance().toObservable().map(new Function<Object, EventMsg>() {
@Override
public EventMsg apply(Object o) throws Exception {
return (EventMsg) o;
}
}).subscribe(new Consumer<EventMsg>() {
@Override
public void accept(EventMsg eventMsg) throws Exception {
if (eventMsg != null) {
mTvContent.setText(eventMsg.getMsg());
}
}
});

第二个页面关键代码

            EventMsg eventMsg = new EventMsg();
            eventMsg.setMsg("来自第二个页面发送过来的数据 --- 修改成功");
            RxBus.getInstance().post(eventMsg);

EventMsg数据类

public class EventMsg {
private String msg;

public String getMsg() {
    return msg;
}

public void setMsg(String msg) {
    this.msg = msg;
}

}

▲ Demo下载地址：https://download.csdn.net/download/donkor_/10312846

About me
Email ：donkor@yeah.net
Android开发交流QQ群 : 537891203

————————————————
原文链接：https://blog.csdn.net/donkor_/article/details/79709366