前两篇文章 放弃RxBus，拥抱RxJava（一）：为什么避免使用EventBus/RxBus ，放弃RxBus，拥抱RxJava（二）：Observable究竟如何封装数据？ 写了一堆理论。看起来并没有什么实际用处，那么今天。我们实战一下，来封装我们需要的数据，并且了解一下各种方式具体的区别。
前言： 很多朋友误会我文章的意思。我写这个系列文章的意思主要是帮助了解一下RxJava的常见用法。而不是使用一下自己或别人封装好的RxBus就觉得自己的项目使用RxJava了。但是这也仅仅是个人口味问题，很多情况下确实RxBus/EventBus会很方便，很刺激，很上瘾。所以从这篇文章开始，我把标题中的"放弃RxBus"去除。
无论在简书,微信平台,GitHub等等分享平台。一个名字上写着 "MVP(MVVM) + RxJava + Retrofit + Dagger2 + ........"这样的名字，再熟悉不过了。然而，大多数情况进去看一下RxJava部分。要么就是简单的把取到的数据用Observable.just()直接传给下一层，要么就是直接使用Retrofit的Adapter来直接获得Observable，而app中其他部分并没有reactive。而且还有很多Observable用法错误，比如冷热不分，连续太多的Map/FlatMap等等。
0. RxBus/Retrofit 足够用了，我为什么要让自己的App 更加的Reactive?
为什么不用RxBus我已经写了两篇文章了，可能由于我不常写文，很多人并没有理解。在这里我再解释一次：EventBus如果是一辆穿梭在所有代码之间的公交车。那么Observable就是穿梭在少许人之间的Uber专车。他比起EventBus有很多优势，比如异常处理，线程切换，强大的操作符等等。你当然可以做出一辆超级Uber来当全局公交车(RxBus)使用，然而这却损失了RxJava本来的许多优势，并且又给自己挖了许多坑。
0.1 一个常见误区，过多的operator
刚开始使用RxJava的时候，我们会觉得operator的链式调用会非常的爽，一个简单的例子：
Observable.just("1", "2", "3", "4", "5", "6", "7")
.map(x -> Integer.valueOf(x))
.map(x -> x * 2)
.map(x -> x + 4)
.filter(x -> x >2)
// and much more operators
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread());
当你只有很少数据的时候，这样当然可以，但是你数据量上来的时候，这就会有很多的overhead。 其实几乎所有的operator都会给你生成一个新的Observable。所以在上面这个例子中，我们在过程中生成了至少7个Observable。然而我们完全可以将中间的.map().map().map().filter合并在一个FlatMap中，减少很多的overhead。
1. Observable.just()的局限性。
- 使用
Observable.just()即使你没有调用subscribe方法。just()括号里面的代码也已经执行了。显然，Observable.just()不适合封装网络数据，因为我们通常不想在subscribe之前做网络请求。
举个例子：
class TestClass{
TestClass(){
System.out.println("I'm created!");
}
}
Observable.just(new TestClass());
这时你运行代码，你就看到确实你的TestClass 已经被创建了：
I/System.out: I'm created!
当然，这个可以简单的用defer()/fromCallable()/create()操作符来是实现只有subscribe只有才加载。
比如：
// use fromCallable
Observable.fromCallable(TestClass::new);
//or
Observable.defer(() -> Observable.just(new TestClass()));
- Observable.just()不够灵活。虽然说设计模式上我们追求 "Minimize Mutability" 但是如果我们的程序越来越 reactive的时候。一个 ObservableJust 往往是不满足需求的。比如之前一定订阅的subscriber。如果数据更新了，你不可以同过ObservableJust 来通知所有的Observable 新数据更新了，需要你的subscriber主动更新。这显然有悖于我们追求的reactive programming。 主动pull数据而不是数据告诉你，我更新了然后再做出反应。
当然ObservableJust在很多情况下，确实不错。如果你不需要监听后续的更新，那么ObservableJust可以满足你的需求。
2. Hot Observable 和 cold Observable
很多人在封装数据的时候，并没有太多考虑冷热的问题，通常情况下并不会出错。因为目前很多开源项目(Demo)里除了RxBus,并没有太多的RxJava的实时情况。然而，当你的App越来越Reactive的时候，冷热便是一个必须考虑的问题。
Hot Observable 意思是如果他开始传输数据，你不主动喊停(dispose()/cancel())，那么他就不会停，一直发射数据，即使他已经没有Subscriber了。而Cold Observable则是subscribe时才会发射数据。
然而，问题来了。我上篇文章讲过，只有subscribeActual方法调用了的时候，Observable发射数据，那为什么Hot Observable没有Subscriber也会发射数据，他把数据发射给谁了呢？我们在解决这个问题之前，先看一下Cold Observable:
2.1 Cold Observable
我们常见的工厂方法提供的都是ColdObservable,包括just(),fromXX,create(),interval(),defer()。 他们的共同点是当你有多个Subscriber的时候，他们的事件是独立的，举个例子：
Observable interval = Observable.interval(1,TimeUnit.SECONDS);
如果我们有两个subscriber,那么他们会各自有自己的计时器，并且互不干扰。效果如下图：
2.2 Hot Observable
不同于Cold Observable, Hot Observable是共享数据的。对于Hot Observable的所有subscriber,他们会在同一时刻收到相同的数据。我们通常使用publish()操作符来将ColdObservable变为Hot。或者我们在RxBus中常常用到的Subjects 也是Hot Observable。
刚刚我们刚刚提出了一个问题，
既然Hot Observable在没有subscriber的时候，还会继续发送数据，那么数据究竟发给谁了呢？
其实Hot Observable其实并没有发送数据，而是他上层的Observable 发送数据给这个hot Observable。不信？我们来分别看一下：
2.2.1 ConnectableObservable
我们在上面的误区中知道了，几乎所有operator都会生成一个新的Observable。publish当然不例外。但是有区别的是，publish会给你一个ConnectableObservable。具体实现类是ObservablePublish。这个Observable的区别是他提供一个connect()方法，如果你调用connect()方法，ConnectableObservable就会开始接收上游Observable的数据。我们来测试一下：
ConnectableObservable interval = Observable.interval(1, TimeUnit.SECONDS).publish();
//connect even when no subscribers
interval.connect();
果然，由于我们subscribe晚了一些。0这个数据没有收到，当我们两个 Subscriber 都dispose的时候，ConnectableObservable 也仍在接受数据，导致我们6这个数据没有接收到。
ConnectableObservable 其实在内部，有一个PublishObserver，他有两个作用。一个是当我们调用 connect()方法时， PublishObserver开始接受上游的数据，我们的例子里便是 Observable.interval(1, TimeUnit.SECONDS) 。所以才能在我们没有调用 subscribe方法时，他也能开始发送数据。第二个作用是 PublishObserver存储所有的下游Subscriber, 也就是我们例子中的Subscriber1 和Subscriber2，在 PublishObserver 每次接到一个上游数据，就会将接收到的结果，依次分发给他存储的所有 Subscribers ,如果下游 Subscriber 调用了 dispose方法，那么他就会在自己的缓存中删除这个 Subscriber，下次接受到上游数据便不会传给这个Subscriber。
那么这时候，有同学应该要问了：
我们可不可以停止从上游接受数据？
我们当然可以。但是从设计的角度，RxJava为了提供链式调用。 connect()方法会返回一个 Disposable 给我们来控制是否继续接受上游的数据。
2.2.2 ConnectableObservable的常用操作符
我们当然不希望每次都手动控制 ConnectableObservable的开关。RxJava给我们提供了一些常用的控制操作符
- refCount()
refCount()可以说是最常用的操作符了。他会把ConnectableObservable变为一个通常的Observable但又保持了HotObservable的特性。也就是说，如果出现第一个Subscriber,他就会自动调用connect()方法，如果他开始接受之后，下游的Subscribers全部dispose,那么他也会停止接受上游的数据。具体看图：
每个 Subscriber 每次都会接受同样的数据，但是当所有 subscriber 都 dispose时候，他也会自动dipose上游的 Observable 。所以我们重新subscribe的时候，又重新从0开始。
这个操作符常用到，RxJava将他和publish合并为一个操作符 ：share()。
- autoConnect()
autoConnect()看名字就知道，他会自动链接，如果你单纯调用autoConnect()，那么，他会在你链接第一个Subscriber的时候调用connect()，或者你调用autoConnect(int Num)，那么他将会再收到Num个subscriber的时候链接。
但是，这个操作符的关键在于，由于我们为了链式调用，autoConnect会返回Observable给你，你不会在返回方法里获得一个Disposable来控制上游的开关。 不过没问题，autoConnect提供了另一种重载方法 ：
autoConnect(int numberOfSubscribers, Consumer<? super Disposable> connection)
他会在这个Consumer传给你 你需要的那个总开关。而且，autoConnect并不会autoDisconnect, 也就是如果他即使没有subscriber了。他也会继续接受数据。 - replay()
replay()方法和publish()一样，会返回一个ConnectableObservable，区别是,replay()会为新的subscriber重放他之前所收到的上游数据，我们再来举个例子：
//only replay 3 values
Observable.interval(1, TimeUnit.SECONDS).replay(3).refCount();
果然，Subscriber2在subscribe时候，立即收到了之前已经错过的三个数据，然后继续接受后面的数据。
但是，这里有几点需要考虑：replay() 会缓存上游发过来的数据，所以并不需要担心重新生成新数据给新的 Subscriber。
- ReplayingShare()
其实ReplayingShare并不能算是ConnectableObservable的一个操作符，他是JakeWhaton的一个开源库，只有百来行。实现的功能是几乎和replay(1).refCount()差不多。但是如果中断 Conncection之后，重新开始subscribe,他仍然会给你一个重放他上一次的结果。 具体看图：
我们看到和刚才的replay不同，即使两个Subscriber都 dispose, 重新开始仍然会接收到我们缓存过的一个数据。
2.3 Subjects
Subjects 作为一个Reactive世界中的特殊存在，他特殊在于他自己既是一个Observable又是一个Observer(Subscriber)。你既可以像普通Observable一样让别的Subscriber来订阅，也可以用Subjects来订阅别人。更方便的是他甚至暴露了OnXX(),方法给你。你直接调用可以通知所有的Subscriber。 这也是RxBus的基础，RxBus几乎离不开Subjects。 蜘蛛侠的老爹告诉我们，力量越大，责任就也大。Subjects也一样。 Subjects因为暴露了OnXX()方法，使得Subjects的数据来源变得难以控制。而且，Subjects一直是HotObservable，我们来看下Subject的OnNext()方法的实现：
@Override
public void onNext(T t) {
if (subscribers.get() == TERMINATED) {
return;
}
if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
for (PublishDisposable<T> s : subscribers.get()) {
s.onNext(t);
}
}
可以看出来Subjects只要调用了OnNext()方法就会立即发送数据。所以，使用时一定要注意Subjects和Subscriber的链接时序问题。具体Subjects的用法我想介绍帖子已经足够多了。这里就不赘述了。
3. 在Android中常见的几种封装和注意事项
1.封装View 的Listener
View 的各种Listener 我们常用create方法来封装，比如OnClickListener:
Observable.create(emitter -> {
button.setOnClickListener(v -> emitter.onNext("I'm Clicked"));
emitter.setCancellable(() -> button.setOnClickListener(null));
});
这里非常关键的一点是一定要设置解除绑定，否则你将持续使用这个会造成内存泄漏。而且最好配合使用share()。否则只有最后一个Subscriber能收到OnClick。当然，如果不考虑方法数的话，推荐配合使用RxBinding。
而且，用create()方法封装Listener适合几乎所有的callback, 并且安全。
2.封装简单的数据源
设想一个场景，我们有一个User类。里面有我们的用户名，头像，各种信息。然而在我们的app中，可能有三四个Fragment/Activity需要根据这个User做出不同的反应。这时我们就可以简单的使用Subject来封装User类。
public class UserRepository {
private User actualUser;
private Subject<User> subject = ReplaySubject.createWithSize(1);
/**
*
*Get User Data from wherever you want Network/Database etc
*/
public Observable<User> getUpdate(){
return subject;
}
public void updateUser(User user){
actualUser = user;
subject.onNext(actualUser);
}
}
如果我们某些模块需要这个User,那么只需要subscribe到这个Repository，如果User有更新，每一个Subscriber都会收到更新后的User并且互相不影响。而且我们使用ReplaySubject,即使有新的Subscriber，也会收到最新的一个Subject。
但是使用的时候一定要注意，因为用的是Subject.所以在onNext方法中一旦出现了error。那么所有的Subscriber都将和这个subject断开了链接。这里也可以用RxRelay代替Subject，简单来说Relay就是一个没有onError和onComplete的Subject。
3.简单的使用concat().first()来处理多来源
Dan Lew在他的博客Loading data from multiple sources with RxJava
中介绍过他这种处理方法，
// Our sources (left as an exercise for the reader)
Observable<Data> memory = ...;
Observable<Data> disk = ...;
Observable<Data> network = ...;
// Retrieve the first source with data
Observable<Data> source = Observable
.concat(memory, disk, network)
.first();
然后在每次做不同请求的时候刷新缓存
Observable<Data> networkWithSave = network.doOnNext(data -> {
saveToDisk(data);
cacheInMemory(data);
});
Observable<Data> diskWithCache = disk.doOnNext(data -> {
cacheInMemory(data);
});
具体也可以看这篇简书，我也不在过多赘述 ：RxJava（八）concat符操作处理多数据源
4.自己继承Observable 手动写subscribeActual()方法
这可能是最灵活的写法？如果你想用RxJava封装自己的库，推荐这种方法封装。因为这样不仅仅可以有效的进行错误处理，并且不会暴露过多逻辑给外面，许多优秀的RxJava相关库都是这样封装，就连RxJava自己也是把一个个的operator封装成一个个不同的Observable。但是这种方法确实要求很高，要做很多考虑，比如异步，多线程冲突，错误处理。对新手不是很推荐。
