为什么在 doOnSubscribe
内调用 subject.onNext(o)
没有任何效果,但是调用 subject.onComplete()
会导致流终止!?
final PublishSubject<Integer> subject = PublishSubject.create();
final Observable<Integer> observable = subject.doOnSubscribe(new Consumer<Disposable>() {
@Override
public void accept(@NonNull Disposable disposable) throws Exception {
System.out.println("disposable = [" + disposable + "]");
subject.onNext(1);
//or
Observable.just(2, 3).subscribe(subject);
}
});
observable.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
System.out.println("d = [" + d.isDisposed() + "]");
}
@Override
public void onNext(Integer integer) {
System.out.println("item = [" + integer + "]");
}
@Override
public void onError(Throwable e) {
System.out.println("e = [" + e + "]");
}
@Override
public void onComplete() {
System.out.println("onComplete");
}
});/*
expected:
disposable = [false]
d = [false]
item = 1
item = 2
item = 3
onComplete
but received :
disposable = [false]
d = [false]
onComplete
*/
最佳答案
订阅 Subject
时在 2.x 中,Disposable
代表连接正在遍历onSubscribe()
特定之前的链条Observer
onNext
变得可见。如果您调用 hasObservers
就可以看到这一点来自onSubscribe
它将返回 false 直到 onSubscribe
实际上返回。
这是 Observable 协议(protocol)所必需的,因为它不允许运行 onSubscribe
和onNext
同时和onSubscribe
必须发生在onNext
之前。如果不遵守此规则,则并发调用 Subject.onNext
会在 Observer.onSubscribe
之前甚至同时运行打电话并找到可能没有准备好的消费者。
自 PublishSubject
不保留任何onNext
未观察到的调用 onNext
元素被丢弃。根据用例,您应该使用 BehaviorSubject
或yourSubject.startWith(initialValue).subscribe()
在任何其他 onNext
之前获取一个值来自Subject
.
关于java - 在 doOnSubscribe 中调用 subject.onNext(),我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/43253399/