RxBus在rxjava2.0 的基本使用

时间:2021-09-05 15:17:00

转载请标明出处: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
    compile 'io.reactivex.rxjava2:rxjava:2.1.8'
    //引入rxAndroid
    compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
    //引入rxJava适配器,方便rxJava与retrofit的结合
    compile 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'
    //引入J神的rxrelay2,出现异常仍然可以处理
    compile '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]:异常处理

▲ 基本使用 二

没有背压处理(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背压

▲ 举个栗子

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

无图无真相,直接看下效果图
RxBus在rxjava2.0 的基本使用

第一个页面关键代码

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
RxBus在rxjava2.0 的基本使用