RxBus基于Rxjava2.0---支持Sticky事件

前言

在项目中我们经常需要两个activity/fragment传递事件，之前我们使用EventBus，而今项目中越来越多的使用Rxjava2，我们就可以使用RxBus来代替EventBus传递事件。 RxBus基于Rxjava，所以使用RxBus时，我们要注意两个情况： ①被观察者在观察者订阅之前发送的事件，观察者无法接收，也就是不支持Sticky事件 ②观察者在接收事件时出现异常会取消订阅 本文就是为解决此问题诞生

RxBus支持Sticky事件

观察者的选择

RxJava提供给开发者4种Subject： PublishSubject，BehaviorSubject ，BehaviorSubject，AsyncSubject。

PublishSubject 是最直接的一个 Subject。当一个数据发射到 PublishSubject 中时，PublishSubject 将立刻把这个数据发射到订阅到该 subject 上的所有 subscriber 中。 ReplaySubject 可以缓存所有发射给他的数据。当一个新的订阅者订阅的时候，缓存的所有数据都会发射给这个订阅者。由于使用了缓存，所以每个订阅者都会收到所以的数据： BehaviorSubject 只保留最后一个值。等同于限制 ReplaySubject 的个数为 1 的情况。在创建的时候可以指定一个初始值，这样可以确保党订阅者订阅的时候可以立刻收到一个值。 AsyncSubject 也缓存最后一个数据。区别是 AsyncSubject 只有当数据发送完成时（onCompleted 调用的时候）才发射这个缓存的最后一个数据。可以使用 AsyncSubject 发射一个数据并立刻结束。

我们考虑使RxBus支持Sticky事件时，还有保证可以和普通事件数据共享，不独立。BehaviorSubject和ReplaySubject貌似可以方便我们实现支持Sticky事件，但是它们可以和普通RxBus的事件数据是分开的，不能保证事件共享，因此这里我们使用PublishSubject，事件数据共享做到了，那么支持Sticky事件呢？我们使用了ConcurrentHashMap<事件类型，事件>保存每个事件的最近事件，这样不仅能实现Sticky特性，最重要的是可以和普通RxBus的事件数据共享了。而且ConcurrentHashMap是一个线程安全的HashMap，效率也很高。

RxBus的实现思路

RxBus的的初始化

private static volatile RxBus mDefaultInstance; private final Subject<Object> mBus; private final Map<Class<?>, Object> mStickyEventMap; public RxBus() { mBus = PublishSubject.create().toSerialized(); mStickyEventMap = new ConcurrentHashMap<>(); } public static RxBus getDefault() { if (mDefaultInstance == null) { synchronized (RxBus.class) { if (mDefaultInstance == null) { mDefaultInstance = new RxBus(); } } } return mDefaultInstance; }

发送Sticky事件

public void postSticky(Object event) { synchronized (mStickyEventMap) { mStickyEventMap.put(event.getClass(), event); } post(event); }

订阅Sticky事件

订阅时调用toObservableSticky方法，先从Map中寻找是否包含该类型的事件，如果没有，直接订阅Observable；如果有，则说明有Sticky事件需要发送，订阅mergeWith整合（可以将多个Observables合并，就好像它们是单个的Observable一样）。

public <T> Observable<T> toObservableSticky(final Class<T> eventType) { synchronized (mStickyEventMap) { Observable<T> observable = mBus.ofType(eventType); final Object event = mStickyEventMap.get(eventType); if (event != null) { return observable.mergeWith(Observable.just(eventType.cast(event))); } else { return observable; } } }

移除Sticky事件

在app退出或者栈底的Activity退出的时候，移除所有的Sticky事件。

@Override protected void onDestroy() { super.onDestroy(); // 移除所有Sticky事件 RxBus.getDefault().removeAllStickyEvents(); }

CompositeDisposable工具类

另外可以写一个RxDisposables方便我们控制对上游事件发送

public class RxDisposables { private static CompositeDisposable mDisposable = new CompositeDisposable(); public static void add(Disposable d) { if (d != null) { mDisposable.add(d); } } public static void remove(Disposable d) { if (d != null) { mDisposable.remove(d); } } public static void clear() { mDisposable.clear(); } public static void dispose() { mDisposable.dispose(); } public static int hasDisposables() { return mDisposable.hashCode(); } }

在订阅事件前，我们先停止之前上游发送事件

RxDisposables.remove(mRxSub);

订阅事件后，将mRxSub添加到CompositeDisposable，方便我们管理上游事件发送

RxDisposables.add(mRxSub);

在所在Activity/Fragment销毁时，停止上游发送事件

@Override public void onDestroy() { super.onDestroy(); RxDisposables.remove(mRxSub); RxDisposables.remove(mRxSubSticky); }

RxBus接收事件出现异常后持续订阅

Rxjava2.0同Rxjava1.0一样，在接收时发生异常，会走到接收出错的方法中，同时会取消订阅，这样的机制有时不满足我们的需求

发送普通事件的实现方式：重新订阅

发送普通事件的思路很简单，既然可以走到接收出错的方法，那么我们在方法里在重新订阅就可以了。我们在map操作中模拟一下异常：

mRxSub = RxBus.getDefault().toObservable(Event.class) .map(new Function<Event, Event>() { @Override public Event apply(Event event) throws Exception { // 发生异常 if (mCheckBox.isChecked()) { throw new RuntimeException("异常"); } return event; } }) .subscribe(new Consumer<Event>() { @Override public void accept(Event event) throws Exception { String str = mTvResult.getText().toString(); mTvResult.setText(TextUtils.isEmpty(str) ? String.valueOf(event.event) : str + ", " + event.event); } }, new Consumer<Throwable>() { @Override public void accept(Throwable throwable) throws Exception { //发生异常，重新订阅 subscribeEvent(); } });

这种方法显然不适合Sticky事件，否则会一直死在那个发生异常的事件上，一直死循环。

发送Sticky事件的实现方式：利用操作符onErrorResumeNext

mRxSubSticky = RxBus.getDefault().toObservableSticky(Event_Sticky.class) .flatMap(new Function<Event_Sticky, ObservableSource<Event_Sticky>>() { @Override public ObservableSource<Event_Sticky> apply(Event_Sticky eventSticky) throws Exception { return Observable.just(eventSticky) .map(new Function<Event_Sticky, Event_Sticky>() { @Override public Event_Sticky apply(Event_Sticky eventSticky) throws Exception { // 发生异常 if (mCheckBox.isChecked()) { throw new RuntimeException("异常"); } return eventSticky; } }) .onErrorResumeNext(new Function<Throwable, ObservableSource<? extends Event_Sticky>>() { @Override public ObservableSource<? extends Event_Sticky> apply(Throwable throwable) throws Exception { return Observable.<Event_Sticky>never(); } }); } }) .subscribe(new Consumer<Event_Sticky>() { @Override public void accept(Event_Sticky eventSticky) throws Exception { String str = mTvResultSticky.getText().toString(); mTvResultSticky.setText(TextUtils.isEmpty(str) ? String.valueOf(eventSticky.event) : str + ", " + eventSticky.event); } }, new Consumer<Throwable>() { @Override public void accept(Throwable throwable) throws Exception { Logger.d(throwable.getMessage()); } });

关键就在于

onErrorResumeNext(Observable.never())

不发射数据并且不会结束订阅。 最后，奉上demo地址： https://github.com/Mr-zhang0101/RxBusTest.git

感谢基于rxjava1.0的rxbus： http://www.jianshu.com/p/71ab00a2677b