Rxjava——使用RxBus替换EventBus

如果你的项目中用到了Rxjava，那么你就完全可以自己去写一个RxBus来替换EventBus，让项目的体积变小。

在网上看了很多人写的RxBus，总感觉有些缺陷或者不能完全替换EventBus的功能。所以，综合别人写的RxBus我自己实现了一个比较完整的RxBus。

首先说说EventBus的功能：

1，注册、反注册、发送数据

2，接收数据（根据类型接收）

3，不同线程接收数据（非ui→非ui、非ui→ui、ui→非ui）

4，观察者只关心发送数据，不关心有多少订阅者

那么我们使用Rxjava也需要来完成这几个功能才算真正的替换。

我们先定义一个RxBus类，并写成单例。

/**
* rxbus的核心类
*/
public class RxBus {

public static RxBus getDefault() {
return RxBusInstance.rxBus;
}
//内部类
private static class RxBusInstance {
private static final RxBus rxBus = new RxBus();
}
// 主题，Subject是非线程安全的
public final Subject bus;

/**
* 单例
* PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者
* 序列化主题
* 将 Subject转换为一个 SerializedSubject ，类中把线程非安全的PublishSubject包装成线程安全的Subject
*/
private RxBus() {
bus = new SerializedSubject<>(PublishSubject.create());
}
}

首相我们得到一个主题（SerializedSubject），这个主题主要的作用就是发送数据。

其中，观察者 Observable是在用户注册的时候得到的对象，它用于接收观察者发出的数据。

订阅对象Subscription用于反注册和判断该对象是否已经被订阅

接下来，我们先写

注册：

/// 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者
public Observable register(Class eventType) {
return bus.ofType(eventType);
}

发送数据：

// 提供了一个新的事件用于发送,这时候的Subject是一个观察者
//ofType将返回一个特定的class类型
public void post(final Object obj) {
bus.onNext(obj);
}

这就是所有的方法，很简单，只有注册和发送数据。

另外还有一些方法例如接收方法，我们可以根据Rxjava的线程规划来写出其他的方式。

但不是在这个类中完成。

接下来我们来看一看它的用法：

我们可以在网络线程等耗时操作中发送数据：

//发送消息
RxBus.getDefault().post(UserInfoBean);

然后在activity或者fragment的onResume方法中注册

@Override
protected void onResume() {
Observable observable=  RxBus.getDefault().register(UserInfoBean.class);//注册用户信息事件
super.onResume();
}

在activity或者fragment的onStop方法中反注册

@Override
protected void onStop() {
if (!observable.isUnsubscribed())
observable.unregister();//如果订阅取消订阅
super.onStop();
}

然后不管在activity的哪个地方都能够去获取数据：

//接收消息，我们不需要关心观察者，只要我们通过observable对象就能得到RxBus发送的数据
Subscription subscriptionUser = observable
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<UserInfoBean>() {
@Override
public void call(UserInfoBeans) {
main_text.append(s.getUserName());
}
});//接收String类型的数据

或者在其他线程

.observeOn(Schedulers.newThread())
.observeOn(Schedulers.io())
//等等

写到这里我想各位看客应该能够理解这个RxBus是如何实现了的吧？

我们来总结一下：

如果我订阅了UserInfoBean的数据类型的话，那么RxBus发送userInfoBean的数据的时候，我就能获取到。不管RxBus写在哪里，不管我具体在哪里需要。

我们只需要从在observable对象中传入UserInfoBean，就可以得到它的数据了。

1，我们在初始化的时候实例化一个主题：

bus = new SerializedSubject<>(PublishSubject.create());

这个主题必须是线程安全的。也就是说它是SerializedSubject而不是Subject。

2，这个主题用于向所有订阅者发送数据：

bus.onNext(obj);

3，注册的时候过滤发送类型。我们传入一个class对象，让这个主题知道，我将要发送这个类型的数据，其他的不会发送。

bus.ofType(cla);

4，于此同时，我们在具体使用到的地方实例化（注册）observable这个观察者实例用于实例化一个订阅对象和告诉订阅者数据来了！

Observable observable = RxBus.getDefault()
.register(String.class);

5，我们从observable对象中获得的订阅对象Subscription，用于获取观察者发送过来的数据

Subscription subscription = observable
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<String>() {
@Override
public void call(String s) {
main_text.append(s);
}
});//接收String类型的数据

5，获得的订阅对象用于判断这个订阅对象是否被订阅以及取消订阅。

//反注册，将注册的subscription对象退订
//是否没订阅
Toast.makeText(this, !subscription.isUnsubscribed() ? "已经订阅" : "未订阅", Toast.LENGTH_LONG).show();
//取消订阅
subscription.unsubscribe();

这个RxBus类就写完了，我们也可用它做EventBus的事情了。

不过……虽然这个类是我在看了他人的RxBus类后改进和改良的。

但是它依然存在问题。

可以看出来，我们做了很多重复的操作。如果我们在一个activity里面需要接受UserInfoBean、IndexInfoBean等等，那么我们就要有几个数据接收就要管理几个观察者和订阅者。

这对于一个优秀的代码来说显然是不够美好的。

不过，为什么不能将他们封装呢？我试过很多方法都无法封装。

因为，我们要依靠它的泛型实例化来确定这个观察者发送的是什么类型的数据。

如果我们封装了观察者，那么很多问题很难解决，至少目前为止我没有解决。

1，如何将具体类型传进去替换掉泛型？

2，一个数据结构对应一个观察者。观察者如何分身多个？

3，一个观察者对应了多个订阅者，如何统一管理和区分注销订阅者？

至少目前为止，只能这么写。如果有哪位看客有更好的封装方法，欢迎探讨！

demo下载