RxJava2.0的简单使用
RxJava2.0---创建被观察者
基于RxJava的事件总线RxBus
RxBus是基于 RxJava 的一种事件总线，相信许多人都了解一些，在RxBus出来之前，相信更多的人是使用EventBus，因为现在 RxJava 太火了，用它几行代码就可以写出事件总线,所以要做到与时俱进。熟悉RxJava的人应该都知道RxJava从1.0版本升级到2.0，做了很多优化的地方 What’s different in 2.0 。下面总结RxJava1.0和 RxJava2.0 版本的 RxBus 写法。
RxJava1.0版本的RxBus
public class RxBus {
private static volatile RxBus defaultInstance;
private final Subject<Object, Object> bus;
// PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者
public RxBus() {
bus = new SerializedSubject<>(PublishSubject.create());
}
// 单例RxBus
public static RxBus getDefault() {
if (defaultInstance == null) {
synchronized (RxBus.class) {
if (defaultInstance == null) {
defaultInstance = new RxBus();
}
}
}
return defaultInstance ;
}
// 发送一个新的事件
public void post (Object o) {
bus.onNext(o);
}
// 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者
public <T> Observable<T> toObservable (Class<T> eventType) {
return bus.ofType(eventType);
}
public Observable<Object> toObserverable() {
return bus;
}
}
RxJava2.0版本的RxBus
在RxJava2.0中把背压和非背压处理分开了，所有出现两种模式。
没有背压处理（Backpressure）的 Rxbus
-
有背压处理的 RxBus
public class RxBus { // // 有背压处理的 RxBus // private final FlowableProcessor<Object> bus; //非背压处理 private final Subject<Object> bus; private static volatile RxBus defaultRxBus; private RxBus() { //非背压处理 bus = PublishSubject.create().toSerialized(); // // 有背压处理的 RxBus // bus = PublishProcessor.create().toSerialized(); } public static RxBus getInstance() { if (null == defaultRxBus) { synchronized (RxBus.class) { if (null == defaultRxBus) { defaultRxBus = new RxBus(); } } } return defaultRxBus; } public void post(Object o) { bus.onNext(o); } public boolean hasObservable() { return bus.hasObservers(); } public Observable<Object> toObservable() { return bus; } /* * 转换为特定类型的Obserbale */ public <T> Observable<T> toObservable(Class<T> type) { return bus.ofType(type); } }
发送通知
RxBus.getInstance().post(new byte[]{0x01, 0x02, 0x03, 0x04});
监听通知
subscribe = RxBus.getInstance().toObservable().subscribe(new Consumer<Object>() {
@Override
public void accept(@NonNull Object o) throws Exception {
if (o instanceof String) {
Toast.makeText(MainActivity.this, "o:" + o, Toast.LENGTH_SHORT).show();
} else if (o instanceof byte[]) {
Toast.makeText(MainActivity.this, "length:" + ((byte[]) o).length, Toast.LENGTH_SHORT).show();
}
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
Toast.makeText(MainActivity.this, throwable.getLocalizedMessage(), Toast.LENGTH_SHORT).show();
}
});
RXBus的基本使用就这么简单，O了。
