[深入RxBus]：支持Sticky事件

2016-07-21 本文已影响9679人 YoKey

RxBus、EventBus因为解耦太彻底，滥用的话，项目可维护性会越来越低；一些简单场景更推荐用回调、Subject来代替事件总线。

实际使用场景，如果RxBus，EventBus二选一，我更倾向于使用EventBus， RxJava专注工作流，EventBus专注事件总线，职责更清晰

有段时间没更了，几个月前，我写过一篇实现简单的RxBus文章：用RxJava实现事件总线。

在实际环境中，你会发现RxBus还是有一些问题的。

你需要RxBus支持Sticky功能。
你会发现在你订阅了某个事件后，在后续接收到该事件时，处理的过程中发生了异常，你可能会发现后续的事件都接收不到了！

我将分2篇文章分别给出其方案，这篇介绍如何实现Sticky，另外一篇介绍RxBus中的异常处理方案:
深入RxBus：［异常处理］

什么是Sticky事件？

在Android开发中，Sticky事件只指事件消费者在事件发布之后才注册的也能接收到该事件的特殊类型。Android中就有这样的实例，也就是Sticky Broadcast，即粘性广播。正常情况下如果发送者发送了某个广播，而接收者在这个广播发送后才注册自己的Receiver，这时接收者便无法接收到刚才的广播，为此Android引入了StickyBroadcast，在广播发送结束后会保存刚刚发送的广播（Intent），这样当接收者注册完Receiver后就可以接收到刚才已经发布的广播。这就使得我们可以预先处理一些事件，让有消费者时再把这些事件投递给消费者。

Subject

我们在实现简单的RxBus时使用了PublishSubject，其实RxJava提供给开发者4种Subject：
PublishSubject，BehaviorSubject ，BehaviorSubject，AsyncSubject。

PublishSubject只会给在订阅者订阅的时间点之后的数据发送给观察者。

BehaviorSubject在订阅者订阅时，会发送其最近发送的数据（如果此时还没有收到任何数据，它会发送一个默认值）。
ReplaySubject在订阅者订阅时，会发送所有的数据给订阅者，无论它们是何时订阅的。
AsyncSubject只在原Observable事件序列完成后，发送最后一个数据，后续如果还有订阅者继续订阅该Subject, 则可以直接接收到最后一个值。

Subject

从上图来看，似乎BehaviorSubject和ReplaySubject具备Sticky的特性。

BehaviorSubject方案

BehaviorSubject似乎完全符合Sticky的定义，但是你发现了它只能保存最近的那个事件。

有这样的场景：如果订阅者A订阅了Event1，订阅者B订阅了Event2，而此时BehaviorSubject事件队列里是［..., Event2, Event1］，当订阅者订阅时，因为保存的是最近的事件：即Event1，所以订阅者B是接收不到Event2的。

解决办法就是：
每个Event类型都各自创建一个对应的BehaviorSubject，这样的话资源开销比较大，并且该Sticky事件总线和普通的RxBus事件总线不能共享，即：普通事件和Sticky事件是独立的，因为普通事件是基于PublishSubject的，暂时放弃该方案！

ReplaySubject方案

ReplaySubject可以保存发送过的所有数据事件。

因为保存了所有的数据事件，所以不管什么类型的Event，我们只要过滤该类型，并让其发送最近的那个Event即可满足Sticky事件了。但是获取最近的对应事件是个难点，因为最符合需求的操作符takeLast()仅在订阅事件结束时（即：onCompleted()）才会发送最后的那个数据事件，而我们的RxBus正常情况下应该是尽量避免其订阅事件结束的。（我没能找到合适的操作符，如果你知道，请告知我）

所以BehaviorSubject也是比较难实现Sticky特性的。

并且，不管是BehaviorSubject还是ReplaySubject，它们还有一个棘手的问题：它们实现的EventBus和普通的RxBus（基于PublishSubject）之间的数据是相互独立的！

总结：BehaviorSubject和BehaviorSubject都不能天然适合Sticky事件......

使用Map实现Sticky

该方法思路是在原来PublishSubject实现的RxBus基础上，使用ConcurrentHashMap<事件类型，事件>保存每个事件的最近事件，不仅能实现Sticky特性，最重要的是可以和普通RxBus的事件数据共享，不独立。

因为我们的RxBus是基于PublishSubject的，而RxJava又有4种Subject，而且其中的BehaviorSubject和ReplaySubject看起来又符合Sticky特性，所以我们可能会钻这个牛角尖，理所当然的认为实现Sticky需要通过其他类型的Subject.... (好吧，我钻进去了...)

这个方案的思路我是根据EventBus的实现想到的，下面是大致流程：

Map的初始化：

   private final Map<Class<?>, Object> mStickyEventMap;

    public RxBus() {
        mBus = new SerializedSubject<>(PublishSubject.create());
        mStickyEventMap = new ConcurrentHashMap<>();
    }

ConcurrentHashMap是一个线程安全的HashMap，采用stripping lock（分离锁），效率比HashTable高很多。

在我们postSticky(Event)时，存入Map中：

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

订阅时toObservableSticky(Class<T> eventType)，先从Map中寻找是否包含该类型的事件，如果没有，则说明没有Sticky事件要发送，直接订阅Subject（此时作为被观察者Observable）；如果有，则说明有Sticky事件需要发送，订阅merge（Subject 和 Sticky事件）。

  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.create(new Observable.OnSubscribe<T>() {
                    @Override
                    public void call(Subscriber<? super T> subscriber) {
                        subscriber.onNext(eventType.cast(event));
                    }
                }));
            } else {
                return observable;
            }
        }
    }

merge操作符：可以将多个Observables合并，就好像它们是单个的Observable一样。

这样，Sticky的核心功能就完成了，使用上和普通RxBus一样，通过postSticky()发送事件，toObservableSticky()订阅事件。

除此之外，我还提供了getStickyEvent(Class<T> eventType),removeStickyEvent(Class<T> eventType),removeAllStickyEvents()方法，供查找、移除对应事件类型的事件、移除全部Sticky事件。

重要的事

在使用Sticky特性时，在不需要某Sticky事件时，通过removeStickyEvent(Class<T> eventType)移除它，最保险的做法是：在主Activity的onDestroy里removeAllStickyEvents()。
因为我们的RxBus是个单例静态对象，再正常退出app时，该对象依然会存在于JVM，除非进程被杀死，这样的话导致StickyMap里的数据依然存在，为了避免该问题，需要在app退出时，清理StickyMap。

// 主Activity（一般是栈底Activity）
@Override
protected void onDestroy() {
    super.onDestroy();
    // 移除所有Sticky事件
    RxBus.getDefault().removeAllStickyEvents();
}

完整代码

下面是支持Sticky的完整RxBus代码:

/**
 * PublishSubject: 只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者
 * <p>
 * Created by YoKeyword on 2015/6/17.
 */
public class RxBus {
    private static volatile RxBus mDefaultInstance;
    private final Subject<Object, Object> mBus;

    private final Map<Class<?>, Object> mStickyEventMap;

    public RxBus() {
        mBus = new SerializedSubject<>(PublishSubject.create());
        mStickyEventMap = new ConcurrentHashMap<>();
    }

    public static RxBus getDefault() {
        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(Class<T> eventType) {
        return mBus.ofType(eventType);
    }

    /**
     * 判断是否有订阅者
     */
    public boolean hasObservers() {
        return mBus.hasObservers();
    }

    public void reset() {
        mDefaultInstance = null;
    }

    /**
     * Stciky 相关
     */

    /**
     * 发送一个新Sticky事件
     */
    public void postSticky(Object event) {
        synchronized (mStickyEventMap) {
            mStickyEventMap.put(event.getClass(), event);
        }
        post(event);
    }

    /**
     * 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者
     */
    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.create(new Observable.OnSubscribe<T>() {
                    @Override
                    public void call(Subscriber<? super T> subscriber) {
                        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();
        }
    }
}

虽然使用了线程安全的ConCurrentHashMap，但是依然大量使用了synchronized，可以发现锁住的是mStickyEventMap对象，这是为了保证对于读、写、查、删同步，即读时不能写，写时不能读...

最后

附上一个Demo：

提供了使用Sticky特性的示例
异常处理的示例：让其在发生异常后，仍能正确接收到后续的Event。

参考源码，传送门

Demo

参考：

ReactiveX: http://reactivex.io/
EventBus: https://github.com/greenrobot/EventBus

[深入RxBus]：支持Sticky事件

什么是Sticky事件？

Subject

BehaviorSubject方案

ReplaySubject方案

使用Map实现Sticky

重要的事

完整代码

最后

参考：

猜你喜欢

热点阅读