这篇文章离上一篇文章有些时日了，概因最难心情大大的不好，非常不爽。
为啥我会专门写一下热发射呢，因为 RxBus 就是使用 RxJava 的热发射(Subject)实现的，但是呢我的出发点不同，我是因为研究了 AAC 的 LivaData 之后才有感而发的，同样是响应式编程，我们如何使用 RxJava 实现构建数据和 UI 之间的通道，做到动态更新数据呢
先来回顾一下 LivaData 的使用，这是介绍 AAC 组件：Android Architecture Components 开发架构 中 demo 的数据流程图
数据流程图
UI 层拿到 modle层抛出的 livedata(Observable) ，在其上注册更新 UI 的方法，然后 modle层在更新数据的时候直接调用 livedata.setValue 方法就可以更新 UI 层的数据了
这其实就是 RxJava 中热发射的典型使用，所以我们真的有必要去学习一下了
先来概念
有热必然有冷，那何为热，何为冷，这点很重要的：
- 冷发射 写
Observable 在收到 Observer 注册时就发送数据，此为冷，我们平时使用的 RxJava 方式都是冷发射 - 热发射
Observable 在收到 Observer 注册时只是建立了相互关系，可以称之为管道。之后 Observable 可以在需要的任何时候发射任意数据。
对于概念我觉得解释的比较明白的是这篇文章里的：Learn RxJava 2 Observables 与 Subscribers
细细品鉴
冷发射经典应用如下
Observable.just(1, 2, 3)
.subscribe(integer -> {
......
});
- 冷发射的数据固定，要不是固定的数据，要不就是一个固定的从远程 IP 获取的数据，是相对写死的
- 冷发射在注册时即开始发射数据，我们不能决定发射数据的时机和地点
热发射经典应用如下
// 创建热发射 Observable
PublishSubject<String> subject = PublishSubject.create();
Disposable disposable = subject.subscribe(s -> show(s));
// 动态发送数据
subject.onNext("响应式编程");
// 中断管道，接触注册关系
disposable.dispose();
- 热发射我们可以决定发射时机，地点，数据，灵活可以实现和 livadata 相同的效果
- 可以解除单个注册
热发射核心 Subject
Subject 这个东西既可以当 Observable 用，上面我们已经看到了，也可以当 Observer 用，接受别的 Observable 的数据
我们先来看看 Subject 的几个实现类，然后才好往下继续
- PublishSubject
- 该Subject不会改变事件的发送顺序。
- 如果在已经发送了一部分事件之后注册的observer，
- 是不会收到之前发送的事件。
private void doPublishSubject() {
//将事件发送到observer，如果先前已经漏掉的事件，不会重新发送到后注册的observer上
PublishSubject<String> publish = PublishSubject.create();
publish.subscribe(new PublishObserver<String>("first"));
publish.onNext("1");
publish.onNext("2");
publish.subscribe(new PublishObserver<String>("seconde"));
publish.onNext("3");
publish.onCompleted();
}
- BehaviorSubject
- 该类有创建时需要一个默认参数，该默认参数会在subject未发送过其他的事件时，向注册的observer发送。
- 注意看代码注释
private void doBehaviorSubject() {
//将事件发送到observer，如果先前已经漏掉的事件，除了最近的一个事件以外，
//其他相关事件不会重新发送到后注册的observer上。所以需要带默认值，
//第一次被observer注册时，observable中没有内容的时候，就会将默认值发给observer
BehaviorSubject<String> behavior = BehaviorSubject.create("创建beahavior时候带的消息");
behavior.subscribe(new SubjectObserver<String>("first"));
behavior.onNext("1");
behavior.onNext("2");
behavior.subscribe(new SubjectObserver<String>("seconde"));
behavior.onNext("3");
behavior.onCompleted();
}
- ReplaySubject
- 将事件发送到observer，无论什么时候注册observer，
- 无论何时通过该observable发射的所有事件，均会发送给新的observer。
private void doReplaySubject() {
//将事件发送到observer，无论什么时候注册observer，
//无论何时通过该observable发射的所有事件，均会发送给新的observer。
ReplaySubject<String> replay = ReplaySubject.create();
replay.subscribe(new SubjectObserver<String>("first"));
replay.onNext("1");
replay.onNext("2");
replay.subscribe(new SubjectObserver<String>("seconde"));
replay.onNext("3");
replay.onCompleted();
}
- AsyncSubject
- 只有当subject调用onComplete方法时，才会将subject中的最后一个事件传递给observer。
- 如果不调用onComplete方法，则不会给observer发送任何事件。
private void doAsyncSubject() {
//只会有当subject调用onComplete方法时，才会将subject中的最后一个事件传递给observer。
//如果不调用onComplete方法，则不会向observer中发送任何事件
AsyncSubject async = AsyncSubject.create();
async.subscribe(new SubjectObserver<String>("first"));
async.onNext("1");
async.onNext("2");
async.onNext("3");
async.onCompleted();
async.subscribe(new SubjectObserver<String>("seconde"));
async.onCompleted();
}
- Subject 作为Observer ，注册到一个冷发射的 Observable 上面
- 注意此时我们只能使用 ReplaySubject
- 因为冷发射注册既发射数据，所有这个 Subject 在注册到冷发射的 Observable 上时就会接受这个冷发射的 Observable 的数据，然后继续向下传递
ReplaySubject<String> subject = ReplaySubject.create();
subject.subscribe(s -> show(s));
Observable<String> observable = Observable.just("AA");
observable.subscribe(subject);
- Subject 作为Observer ，注册到一个热发射的 Observable 上面
- 这是既没有类型限制了
- 我们添加一个变换进来，要不然没啥意义，这个其实和 livadata 天剑 transform 一个思路
PublishSubject<String> subject1 = PublishSubject.create();
PublishSubject<String> subject2 = PublishSubject.create();
subject1.map(new Function<String, String>() {
@Override
public String apply(String s) throws Exception {
return s + "_经过2的修改了";
}
}).subscribe(subject2);
subject2.subscribe(s -> show(s));
subject1.onNext("响应式编程");
ConnnectableObservbale
我们可以把一个冷发射转为热发射，使用 publish
ConnectableObservable<String> source = Observable
.just("Alpha","Beta","Delta","Gamma","Epsilon")
.publish();
source.subscribe(s -> System.out.println("observer1 RECEIVED: " + s));
source.map(String::length)
.subscribe(i -> System.out.println("observer2 RECEIVED: " + i));
//发射！
source.connect();
不过我觉得这个 ConnnectableObservbale 没啥发用，写死数据的热发射没太大应用价值。
代码特征
PublishSubject 我想大家肯定有一些疑问这里我测试过直接上答案
1. PublishSubject 变换不会丢失通道的特性，并且一样可以多注册
val subject = PublishSubject.create<String>()
val observable = subject.map {
return@map "BB"
}
observable.subscribe {
Log.d("AA", "收到数据1：$it")
}
observable.subscribe {
Log.d("AA", "收到数据2：$it")
}
subject.onNext("AA")
PublishSubject 变换之后虽然我们拿到的是 Observable，但是 PublishSubject 特性不会丢失，为啥？因为数据源头不会像 Observable 一样由注册的就触发数据，这个 Observable 的源头还是 PublishSubject ，Observable 求到的只是衔接作用
