最近再看一个项目，但是那个项目里面的Rxjava是1.x版本的，由于最近又有一个项目要开始了，在封装各种基类，所以我准备将项目中的Rxbus用Rxjava2.x来修改一下继续用，由于2.x的Rxjava进行了代码的重构，所以在这里写下我的一些收集与写法

什么是Rxbus

先说说Rxjava

官方解释：

RxJava is a Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences.

RxJava是Reactive Extensions的Java VM实现：用于通过使用observable序列来组合异步和基于事件的程序的库。

那我们的Rxbus呢？

RxBus：处理应用程序间各个组件的通信，或者组件与组建之间的数据传递。

其实说到底，RxBus学的是一种思路，而并不是给你一个现成的库，然后直接去调用。需要的是自己来封装。

Rxbus的简单实现

导包

compile 'io.reactivex.rxjava2:rxjava:2.1.2'

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

Rxbus类

public class RxBus {

// 主题

private final Subject bus;

// PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者

private RxBus() {

bus = PublishSubject.create().toSerialized();

}

// 静态内部类(单例模式的内部类实现方法)

private static class SingletonHolder {

public final static RxBus sInstance = new RxBus();

}

// 单例RxBus

public static RxBus getDefault() {

return SingletonHolder.sInstance;

}

/**

* 提供了一个新的事件

* 发布

* @param o

*/

public void post(Object o) {

bus.onNext(o);

}

// 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者

public Observable toObserverable(Class eventType) {

return bus.ofType(eventType); //判断接收事件类型

}

}

不侮辱大家智商，配上注释应该大家都懂了，就不解释了

Rxjava1.x -> 2.x问题

如果大家有兴趣的话，最好还是阅读源码。

1. 修改EventThread.class，删除Schedulers.immediate()相关

因为在2.0中删除了Schedulers.immediate()这个线程的切换

2. //存放订阅者信息

private Map subscriptions = new HashMap<>();

修改为：

private Map subscriptions = new HashMap<>();

3. CompositeSubscription.unsubscribe();

修改为

CompositeDisposable.dispose();

4. 在2.0中增加了Flowable 这样就把 backpressure 的问题放到了Flowable中来处理，

而Observable 不对backpressure进行处理了。但是使用Flowable还是要注意对backpressure

的处理，不然还是会出现以前的问题。

5. Subscription subscription = toObservable(sub.tag(), cla);

修改为：

Disposable disposable = tObservable(sub.tag(), cla);

6. SerializedSubject 已经变为非public类,可以通过

bus = PublishSubject.create().toSerialized();

的方式获取线程安全 的对象。

7. private final PublishSubject bus = PublishSubject.create();

final Subject subject = bus.toSerialized();

修改为：

bus = PublishSubject.create().toSerialized();

toSerialized() 方法中其实就是之前序列化安全对象的写法。而SerializedSubject类已经变成了非public的

8.//1.X public class AppBaseActivity extends AppCompatActivity {

...

private CompositeSubscription mCompositeSubscription;

protected void addSubscription(Subscription subscription) {

if (null == mCompositeSubscription) {

mCompositeSubscription = new CompositeSubscription();

}

mCompositeSubscription.add(subscription);

}

@Override

protected void onDestroy() {

if (null != mCompositeSubscription) {

mCompositeSubscription.unsubscribe();

}

super.onDestroy();

}

...

}

//2.X

public class AppBaseActivity extends AppCompatActivity {

...

private CompositeDisposable mCompositeDisposable;

protected void addDisposable(Disposable disposable) {

if (null == mCompositeDisposable) {

mCompositeDisposable = new CompositeDisposable();

}

mCompositeDisposable.add(disposable);

}

@Override

protected void onDestroy() {

if (null != mCompositeDisposable) {

mCompositeDisposable.clear();

}

super.onDestroy();

}

...

}