Android 打造RxBus2.x的全面详解

前言

在Android EventBus3.x的使用详解一文中我们全面讲解了使用EvenBus3.x解决进程/界面通信。
本文将介绍另一个东西—>使用RxBus来实现进程/界面通信。
阅读本文，你需要提前了解RxJava，可以查看：
给 Android 开发者的 RxJava 详解
Android响应式编程框架—RxJava&RxAndroid2.0使用笔记

RxBus简介

什么是RxBus？

RxBus 名字看起来像一个库，但它并不是一个库，而是一种模式，它的思想是使用RxJava来实现了EventBus ，而让你不再需要使用 Otto 或者 GreenRobot 的 EventBus。------ 给 Android 开发者的 RxJava 详解

RxBus2.x就是基于RxJava2.x封装实现的类。

为什么要使用RxBus？

如上所说，我们可以通过封装RxJava实现EventBus。
随着RxJava在Android项目中的普及, 我们完全可以使用RxBus代替EventBus，减少库的引入，增加系统的稳定性。
EventBus虽然使用方便，但是在事件的生命周期的处理上需要我们利用订阅者的生命周期去注册和取消注册，这个部分还是略有麻烦之处。
而我们可以结合使用RxLifecycle来配置，简化这一步骤。
结合RxJava的强大能力和RxAndroid的进程调度，RxBus有更多更强大的功能。

RxBus实现

Gradle

    implementation 'io.reactivex.rxjava2:rxjava:2.1.17' //RxJavaimplementation 'io.reactivex.rxjava2:rxandroid:2.0.2' //RxAndroid

封装RxJava实现RxBus

Subject有两种用途:

做为observable向其他的observable发送事件
做为observer接收其他的observable发送的事件

我们要实现：事件总线、事件发布者以及事件订阅者。首先Subject既可以作为被观察者发送事件，也可以作为观察者接收事件，而RxJava内部的响应式的支持实现了事件总线的功能。
可以使用PublishSubject.create().toSerialized();生成一个Subject对象。
如下代码就实现了最基本的RxBus。

public class RxBus {private volatile static RxBus mDefaultInstance;private final Subject<Object> mBus;private RxBus() {mBus = PublishSubject.create().toSerialized();}public static RxBus getInstance() {if (mDefaultInstance == null) {synchronized (RxBus.class) {if (mDefaultInstance == null) {mDefaultInstance = new RxBus();}}}return mDefaultInstance;}/*** 发送事件*/public void post(Object event) {mBus.onNext(event);}/*** 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者*/public <T> Observable<T> toObservable(final Class<T> eventType) {return mBus.ofType(eventType);}/*** 判断是否有订阅者*/public boolean hasObservers() {return mBus.hasObservers();}public void reset() {mDefaultInstance = null;}}

使用RxBus

事件实体

public class MsgEvent {private String msg;public MsgEvent(String msg) {this.msg = msg;}public String getMsg() {return msg;}public void setMsg(String msg) {this.msg = msg;}
}

订阅事件

RxBus.getInstance().toObservable(MsgEvent.class).subscribe(new Observer<MsgEvent>() {@Overridepublic void onSubscribe(Disposable d) {}@Overridepublic void onNext(MsgEvent msgEvent) {//处理事件}@Overridepublic void onError(Throwable e) {}@Overridepublic void onComplete() {}});

发送事件


RxBus.getInstance().post(new MsgEvent("Java"));

与EventBus对比，除了订阅事件的消息处理，使用基本一样。
至于EventBus的线程模型，我们完全可以使用RxJava 的线程控制Scheduler来实现，具体可以参考上面RxJava的使用两篇文章。

RxBus内存泄漏

使用RxJava发布一个订阅后，当页面被finish，此时订阅逻辑还未完成，如果没有及时取消订阅，就会导致Activity/Fragment无法被回收，从而引发内存泄漏。
EventBus为了解决这个问题，要求我们根据订阅者的生命周期注册和取消注册。所以在RxBus中我们也可以这样操作。

简单处理

使用一个CompositeDisposable存储当前所有的订阅，然后再onDestroy()中将其dispose()。

private CompositeDisposable compositeDisposable;@Overrideprotected void onCreate(Bundle savedInstanceState) {compositeDisposable = new CompositeDisposable();RxBus.getInstance().toObservable(MsgEvent.class).subscribe(new Observer<MsgEvent>() {@Overridepublic void onSubscribe(Disposable d) {compositeDisposable.add(d);}@Overridepublic void onNext(MsgEvent msgEvent) {//事件处理}@Overridepublic void onError(Throwable e) {}@Overridepublic void onComplete() {}});}@Overridepublic void onDestroy() {super.onDestroy();unbinder.unbind();if (compositeDisposable!=null && !compositeDisposable.isDisposed()){compositeDisposable.dispose();}}

RxLifecycle

Rxlifecycle 是trello开发的用于解决RxJava引起的内存泄漏的开源框架。
GitHub地址：https://github.com/trello/RxLifecycle
该框架为了适应不同的场景,开发了不同的版本，具体的可以查看GitHub文档:

// RxLifecycle基础库
implementation 'com.trello.rxlifecycle2:rxlifecycle:2.2.2'// Android使用的库，可以绑定特定的生命周期
//需继承RxActvivty使用
implementation 'com.trello.rxlifecycle2:rxlifecycle-android:2.2.2'// Android组件库，里面定义了例如RxAppCompatActivity、RxFragment之类的Android组件
// 须继承RxAppCompatActivity、RxFragment使用
implementation 'com.trello.rxlifecycle2:rxlifecycle-components:2.2.2'// 预先编写的支持首选项片段，将其子类化为提供者
implementation 'com.trello.rxlifecycle2:rxlifecycle-components-preference:2.2.2'// Android使用的库，须继承NaviActivity使用
implementation 'com.trello.rxlifecycle2:rxlifecycle-navi:2.2.2'// 使用Android生命周期作为提供者
//无需继承，任何有声明周期的组件都可以直接使用
implementation 'com.trello.rxlifecycle2:rxlifecycle-android-lifecycle:2.2.2'// Kotlin语法
implementation 'com.trello.rxlifecycle2:rxlifecycle-kotlin:2.2.2'//在Kotlin语法，使用Android生命周期
implementation 'com.trello.rxlifecycle2:rxlifecycle-android-lifecycle-kotlin:2.2.2'

如果我们自己有BaseActivity，所以不能继承RxActvivty，RxAppCompatActivity、RxFragment，NaviActivity。
为了保持代码的灵活性，我们使用:

// 使用Android生命周期作为提供者
implementation 'com.trello.rxlifecycle2:rxlifecycle-android-lifecycle:2.2.2'

修改RxBus，添加如下代码：
使用compose(this.bindToLifecycle())方法绑定Activity的生命周期，在onStart方法中绑定，在onStop方法被调用后就会解除绑定，以此类推。
如果在onPause/onStop方法中绑定，那么就会在它的下一个生命周期方法（onStop/onDestory）被调用后解除绑定。

绑定	销毁
onCreate	onDestory
onStart	onStop
onResum	onPause
onPause	onStop
onStop	onDestory

 /*** 使用Rxlifecycle解决RxJava引起的内存泄漏*/public <T> Observable<T> toObservable(LifecycleOwner owner, final Class<T> eventType) {LifecycleProvider<Lifecycle.Event> provider = AndroidLifecycle.createLifecycleProvider(owner);return mBus.ofType(eventType).compose(provider.<T>bindToLifecycle());}

订阅事件时，添加上下文this即可绑定生命周期，自动取消订阅：

RxBus.getInstance().toObservable(this,MsgEvent.class).subscribe(new Consumer<MsgEvent>() {@Overridepublic void accept(MsgEvent msgEvent) throws Exception {//处理事件}});

RxBus的粘性事件

EventBus有粘性事件，RxBus也可以实现。
修改RxBus修改&添加如下代码：

private final Map<Class<?>, Object> mStickyEventMap;private RxBus() {mBus = PublishSubject.create().toSerialized();mStickyEventMap = new ConcurrentHashMap<>();}/*** 发送一个新Sticky事件*/public void postSticky(Object event) {synchronized (mStickyEventMap) {mStickyEventMap.put(event.getClass(), event);}post(event);}/*** 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者* 使用Rxlifecycle解决RxJava引起的内存泄漏*/public <T> Observable<T> toObservableSticky(LifecycleOwner owner,final Class<T> eventType) {synchronized (mStickyEventMap) {LifecycleProvider<Lifecycle.Event> provider = AndroidLifecycle.createLifecycleProvider(owner);Observable<T> observable = mBus.ofType(eventType).compose(provider.<T>bindToLifecycle());final Object event = mStickyEventMap.get(eventType);if (event != null) {return observable.mergeWith(Observable.create(new ObservableOnSubscribe<T>() {@Overridepublic void subscribe(ObservableEmitter<T> subscriber) throws Exception {subscriber.onNext(eventType.cast(event));}}));} else {return observable;}}}/*** 根据eventType获取Sticky事件*/public <T> T getStickyEvent(Class<T> eventType) {synchronized (mStickyEventMap) {return eventType.cast(mStickyEventMap.get(eventType));}}/*** 移除指定eventType的Sticky事件*/public <T> T removeStickyEvent(Class<T> eventType) {synchronized (mStickyEventMap) {return eventType.cast(mStickyEventMap.remove(eventType));}}/*** 移除所有的Sticky事件*/public void removeAllStickyEvents() {synchronized (mStickyEventMap) {mStickyEventMap.clear();}}

2019.11.13更新

实际上使用时发现：
如果组件没有继承RxComponents(如RxActivity)则不会按照预期的相反的生命周期事件中结束序列。
因此建议如不继承RxComponents就使用provider.bindUntilEvent来指明在哪个生命周期事件中结束序列。

 AndroidLifecycle.createLifecycleProvider(LifecycleOwner owner).<T>bindUntilEvent(Lifecycle.Event event)

LifecycleOwner owner 生命周期所有者，传上下文this。
Lifecycle.Event event 在指定的生命周期事件中结束序列(RxJava)。

Lifecycle.Event	对应生命周期
ON_CREATE	在onCreate之后
ON_START	在onStart之后
ON_RESUME	在onResume之后
ON_PAUSE	再onPause之前
ON_STOP	在onStop之前
ON_DESTROY	在onDestory之前
ON_ANY	match all events

消息事件实体

class MsgEvent(var msg: String)

普通消息事件

接收消息事件

//Lifecycle.Event.ON_DESTROY
RxBus.getInstance().toObservable(this, MsgEvent::class.java).subscribe { msg ->//处理消息}

或者

RxBus.getInstance().toObservable(this, MsgEvent::class.java, Lifecycle.Event.ON_PAUSE).subscribe { msg ->//处理消息}

发布消息事件

RxBus.getInstance().post(MsgEvent("Java"));

粘性消息事件（Sticky）

用于解决先发布事件，然后再订阅事件的情况。

接收粘性消息事件

//Lifecycle.Event.ON_DESTROY
RxBus.getInstance().toObservableSticky(this, MsgEvent::class.java).subscribe { msg ->//处理消息}

或者

RxBus.getInstance().toObservableSticky(this, MsgEvent::class.java, Lifecycle.Event.ON_PAUSE).subscribe { msg ->//处理消息}

发送粘性消息事件

  RxBus.getInstance().postSticky(MsgEvent("Java"));

Demo&代码

https://github.com/DeMonLiu623/DeMon-RxBus