Cold Observable 和 Hot Observable
Observable的分类
Observable 有 Cold 和 Hot 之分。html
Hot Observable 不管有没有 Subscriber 订阅，事件始终都会发生。当 Hot Observable 有多个订阅者时，Hot Observable 与订阅者们的关系是一对多的关系，能够与多个订阅者共享信息。java
然而，Cold Observable 只有 Subscriber 订阅时，才开始执行发射数据流的代码。而且 Cold Observable 和 Subscriber 只能是一对一的关系，当有多个不一样的订阅者时，消息是从新完整发送的。也就是说对 Cold Observable 而言，有多个Subscriber的时候，他们各自的事件是独立的。react
若是上面的解释有点枯燥的话，那么下面会更加形象地说明 Cold 和 Hot 的区别：安全
Think of a hot Observable as a radio station. All of the listeners that are listening to it at this moment listen to the same song.
A cold Observable is a music CD. Many people can buy it and listen to it independently.
by Nickolay Tsvetinov服务器
Cold Observable
Observable 的 just、creat、range、fromXXX 等操做符都能生成Cold Observable。网络
Consumer<Long> subscriber1 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println("subscriber1: "+aLong);
}
};
Consumer<Long> subscriber2 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println(" subscriber2: "+aLong);
}
};
Observable<Long> observable = Observable.create(new ObservableOnSubscribe<Long>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
Observable.interval(10, TimeUnit.MILLISECONDS,Schedulers.computation())
.take(Integer.MAX_VALUE)
.subscribe(e::onNext);
}
}).observeOn(Schedulers.newThread());
observable.subscribe(subscriber1);
observable.subscribe(subscriber2);
try {
Thread.sleep(100L);
} catch (InterruptedException e) {
e.printStackTrace();
}复制代码
执行结果：并发
subscriber1: 0
subscriber2: 0
subscriber1: 1
subscriber2: 1
subscriber1: 2
subscriber2: 2
subscriber2: 3
subscriber1: 3
subscriber1: 4
subscriber2: 4
subscriber2: 5
subscriber1: 5
subscriber1: 6
subscriber2: 6
subscriber1: 7
subscriber2: 7
subscriber1: 8
subscriber2: 8
subscriber1: 9
subscriber2: 9复制代码
能够看出，subscriber1 和 subscriber2 的结果并不必定是相同的，两者是彻底独立的。app
尽管 Cold Observable 很好，可是对于某些事件不肯定什么时候发生以及不肯定 Observable 发射的元素数量，那还得使用 Hot Observable。好比：UI交互的事件、网络环境的变化、地理位置的变化、服务器推送消息的到达等等。ide
Cold Observable 如何转换成 Hot Observable？
1. 使用publish，生成 ConnectableObservable
使用 publish 操做符，可让 Cold Observable 转换成 Hot Observable。它将原先的 Observable 转换成 ConnectableObservable。this
来看看刚才的例子：
Consumer<Long> subscriber1 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println("subscriber1: "+aLong);
}
};
Consumer<Long> subscriber2 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println(" subscriber2: "+aLong);
}
};
Consumer<Long> subscriber3 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println(" subscriber3: "+aLong);
}
};
ConnectableObservable<Long> observable = Observable.create(new ObservableOnSubscribe<Long>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
Observable.interval(10, TimeUnit.MILLISECONDS,Schedulers.computation())
.take(Integer.MAX_VALUE)
.subscribe(e::onNext);
}
}).observeOn(Schedulers.newThread()).publish();
observable.connect();
observable.subscribe(subscriber1);
observable.subscribe(subscriber2);
try {
Thread.sleep(20L);
} catch (InterruptedException e) {
e.printStackTrace();
}
observable.subscribe(subscriber3);
try {
Thread.sleep(100L);
} catch (InterruptedException e) {
e.printStackTrace();
}复制代码
注意，生成的 ConnectableObservable 须要调用connect()才能真正执行。
执行结果：
subscriber1: 0
subscriber2: 0
subscriber1: 1
subscriber2: 1
subscriber1: 2
subscriber2: 2
subscriber3: 2
subscriber1: 3
subscriber2: 3
subscriber3: 3
subscriber1: 4
subscriber2: 4
subscriber3: 4
subscriber1: 5
subscriber2: 5
subscriber3: 5
subscriber1: 6
subscriber2: 6
subscriber3: 6
subscriber1: 7
subscriber2: 7
subscriber3: 7
subscriber1: 8
subscriber2: 8
subscriber3: 8
subscriber1: 9
subscriber2: 9
subscriber3: 9
subscriber1: 10
subscriber2: 10
subscriber3: 10
subscriber1: 11
subscriber2: 11
subscriber3: 11复制代码
能够看到，多个订阅的 Subscriber 共享同一事件。
在这里，ConnectableObservable 是线程安全的。
2. 使用Subject/Processor
Subject 和 Processor 的做用是相同的。Processor 是 RxJava2.x 新增的类，继承自 Flowable 支持背压控制。而 Subject 则不支持背压控制。
Consumer<Long> subscriber1 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println("subscriber1: "+aLong);
}
};
Consumer<Long> subscriber2 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println(" subscriber2: "+aLong);
}
};
Consumer<Long> subscriber3 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println(" subscriber3: "+aLong);
}
};
Observable<Long> observable = Observable.create(new ObservableOnSubscribe<Long>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
Observable.interval(10, TimeUnit.MILLISECONDS,Schedulers.computation())
.take(Integer.MAX_VALUE)
.subscribe(e::onNext);
}
}).observeOn(Schedulers.newThread());
PublishSubject<Long> subject = PublishSubject.create();
observable.subscribe(subject);
subject.subscribe(subscriber1);
subject.subscribe(subscriber2);
try {
Thread.sleep(20L);
} catch (InterruptedException e) {
e.printStackTrace();
}
subject.subscribe(subscriber3);
try {
Thread.sleep(100L);
} catch (InterruptedException e) {
e.printStackTrace();
}复制代码
执行结果跟上面使用 publish 操做符是同样的。
Subject 既是 Observable 又是 Observer(Subscriber)。这一点能够从 Subject 的源码上看到。
import io.reactivex.*;
import io.reactivex.annotations.*;
/** * Represents an Observer and an Observable at the same time, allowing * multicasting events from a single source to multiple child Subscribers. * <p>All methods except the onSubscribe, onNext, onError and onComplete are thread-safe. * Use {@link #toSerialized()} to make these methods thread-safe as well. * * @param <T> the item value type */
public abstract class Subject<T> extends Observable<T> implements Observer<T> {
/** * Returns true if the subject has any Observers. * <p>The method is thread-safe. * @return true if the subject has any Observers */
public abstract boolean hasObservers();
/** * Returns true if the subject has reached a terminal state through an error event. * <p>The method is thread-safe. * @return true if the subject has reached a terminal state through an error event * @see #getThrowable() * &see {@link #hasComplete()} */
public abstract boolean hasThrowable();
/** * Returns true if the subject has reached a terminal state through a complete event. * <p>The method is thread-safe. * @return true if the subject has reached a terminal state through a complete event * @see #hasThrowable() */
public abstract boolean hasComplete();
/** * Returns the error that caused the Subject to terminate or null if the Subject * hasn't terminated yet. * <p>The method is thread-safe. * @return the error that caused the Subject to terminate or null if the Subject * hasn't terminated yet */
@Nullable
public abstract Throwable getThrowable();
/** * Wraps this Subject and serializes the calls to the onSubscribe, onNext, onError and * onComplete methods, making them thread-safe. * <p>The method is thread-safe. * @return the wrapped and serialized subject */
@NonNull
public final Subject<T> toSerialized() {
if (this instanceof SerializedSubject) {
return this;
}
return new SerializedSubject<T>(this);
}
}复制代码
当 Subject 做为 Subscriber 时，它能够订阅目标 Cold Observable 使对方开始发送事件。同时它又做为Observable 转发或者发送新的事件，让 Cold Observable 借助 Subject 转换为 Hot Observable。
注意，Subject 并非线程安全的，若是想要其线程安全须要调用toSerialized()方法。(在RxJava1.x的时代还能够用 SerializedSubject 代替 Subject，可是在RxJava2.x之后SerializedSubject再也不是一个public class）
然而，不少基于 EventBus 改造的 RxBus 并无这么作，包括我之前也写过这样的 RxBus :( 。这样的作法是很是危险的，由于会遇到并发的状况。
Hot Observable 如何转换成 Cold Observable？
1. ConnectableObservable的refCount操做符
reactivex官网的解释是
make a Connectable Observable behave like an ordinary Observable
RefCount操做符把从一个可链接的 Observable 链接和断开的过程自动化了。它操做一个可链接的Observable，返回一个普通的Observable。当第一个订阅者订阅这个Observable时，RefCount链接到下层的可链接Observable。RefCount跟踪有多少个观察者订阅它，直到最后一个观察者完成才断开与下层可链接Observable的链接。
若是全部的订阅者都取消订阅了，则数据流中止。若是从新订阅则从新开始数据流。
Consumer<Long> subscriber1 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println("subscriber1: "+aLong);
}
};
Consumer<Long> subscriber2 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println(" subscriber2: "+aLong);
}
};
ConnectableObservable<Long> connectableObservable = Observable.create(new ObservableOnSubscribe<Long>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
Observable.interval(10, TimeUnit.MILLISECONDS,Schedulers.computation())
.take(Integer.MAX_VALUE)
.subscribe(e::onNext);
}
}).observeOn(Schedulers.newThread()).publish();
connectableObservable.connect();
Observable<Long> observable = connectableObservable.refCount();
Disposable disposable1 = observable.subscribe(subscriber1);
Disposable disposable2 = observable.subscribe(subscriber2);
try {
Thread.sleep(20L);
} catch (InterruptedException e) {
e.printStackTrace();
}
disposable1.dispose();
disposable2.dispose();
System.out.println("从新开始数据流");
disposable1 = observable.subscribe(subscriber1);
disposable2 = observable.subscribe(subscriber2);
try {
Thread.sleep(20L);
} catch (InterruptedException e) {
e.printStackTrace();
}复制代码
执行结果：
subscriber1: 0
subscriber2: 0
subscriber1: 1
subscriber2: 1
从新开始数据流
subscriber1: 0
subscriber2: 0
subscriber1: 1
subscriber2: 1复制代码
若是不是全部的订阅者都取消了订阅，只取消了部分。部分的订阅者从新开始订阅，则不会从头开始数据流。
Consumer<Long> subscriber1 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println("subscriber1: "+aLong);
}
};
Consumer<Long> subscriber2 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println(" subscriber2: "+aLong);
}
};
Consumer<Long> subscriber3 = new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
System.out.println(" subscriber3: "+aLong);
}
};
ConnectableObservable<Long> connectableObservable = Observable.create(new ObservableOnSubscribe<Long>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Long> e) throws Exception {
Observable.interval(10, TimeUnit.MILLISECONDS,Schedulers.computation())
.take(Integer.MAX_VALUE)
.subscribe(e::onNext);
}
}).observeOn(Schedulers.newThread()).publish();
connectableObservable.connect();
Observable<Long> observable = connectableObservable.refCount();
Disposable disposable1 = observable.subscribe(subscriber1);
Disposable disposable2 = observable.subscribe(subscriber2);
observable.subscribe(subscriber3);
try {
Thread.sleep(20L);
} catch (InterruptedException e) {
e.printStackTrace();
}
disposable1.dispose();
disposable2.dispose();
System.out.println("subscriber一、subscriber2 从新订阅");
disposable1 = observable.subscribe(subscriber1);
disposable2 = observable.subscribe(subscriber2);
try {
Thread.sleep(20L);
} catch (InterruptedException e) {
e.printStackTrace();
}复制代码
执行结果：
subscriber1: 0
subscriber2: 0
subscriber3: 0
subscriber1: 1
subscriber2: 1
subscriber3: 1
subscriber一、subscriber2 从新订阅
subscriber3: 2
subscriber1: 2
subscriber2: 2
subscriber3: 3
subscriber1: 3
subscriber2: 3
subscriber3: 4
subscriber1: 4
subscriber2: 4复制代码
在这里，subscriber1和subscriber2先取消了订阅，subscriber3并无取消订阅。以后，subscriber1和subscriber2又从新订阅。最终subscriber一、subscriber二、subscriber3的值保持一致。
2. Observable的share操做符
share操做符封装了publish().refCount()调用，能够看其源码。
/** * Returns a new {@link ObservableSource} that multicasts (shares) the original {@link ObservableSource}. As long as * there is at least one {@link Observer} this {@link ObservableSource} will be subscribed and emitting data. * When all subscribers have disposed it will dispose the source {@link ObservableSource}. * <p> * This is an alias for {@link #publish()}.{@link ConnectableObservable#refCount()}. * <p> *  * <dl> * <dt><b>Scheduler:</b></dt> * <dd>{@code share} does not operate by default on a particular {@link Scheduler}.</dd> * </dl> * * @return an {@code ObservableSource} that upon connection causes the source {@code ObservableSource} to emit items * to its {@link Observer}s * @see <a href="http://reactivex.io/documentation/operators/refcount.html">ReactiveX operators documentation: RefCount</a> */
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> share() {
return publish().refCount();
}复制代码
总结
理解了 Hot Observable 和 Cold Observable 的区别才可以写出更好Rx代码。同理，也能理解Hot & Cold Flowable。再者，在其余语言的Rx版本中包括 RxSwift、RxJS 等也存在 Hot Observable 和 Cold Observable 这样的概念。
相关文章
- 1. [RxJS] Hot Observable, by .share()
- 2. Observable
- 3. Angular-Observable和RxJS
- 4. [rxjs] Demystifying Cold and Hot Observables in RxJS
- 5. Rxjs-Observable
- 6. Observer Observable
- 7. [译] 经过构建 Observable 来学习 Observable
- 8. RXJS Observable的冷，热和Subject
- 9. Rxjava2 Observable源码浅析
- 10. BehaviorSubject vs Observable？
- 更多相关文章...
- • XLink 和 XPointer 语法 - XLink 和 XPointer 教程
- • Kotlin 类和对象 - Kotlin 教程
- • RxJava操作符（九）Connectable Observable Operators
- • IntelliJ IDEA 代码格式化配置和快捷键
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. [RxJS] Hot Observable, by .share()
- 2. Observable
- 3. Angular-Observable和RxJS
- 4. [rxjs] Demystifying Cold and Hot Observables in RxJS
- 5. Rxjs-Observable
- 6. Observer Observable
- 7. [译] 经过构建 Observable 来学习 Observable
- 8. RXJS Observable的冷，热和Subject
- 9. Rxjava2 Observable源码浅析
- 10. BehaviorSubject vs Observable？