在 doOnSubscribe 中调用 subject.onNext()答案

【问题标题】：Calling subject.onNext() inside doOnSubscribe在 doOnSubscribe 中调用 subject.onNext()
【发布时间】：2017-04-06 19:19:41
【问题描述】：

为什么在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
    */

【问题讨论】：

标签： java rx-java2

【解决方案1】：

在 2.x 中订阅 Subject 时，表示连接的 Disposable 在特定的 Observer 对 onNext 可见之前遍历 onSubscribe() 链。如果您从onSubscribe 调用hasObservers，您可以看到这一点，在onSubscribe 实际返回之前，它将返回false。

这是 Observable 协议所要求的，因为它不允许同时运行 onSubscribe 和 onNext 并且 onSubscribe 必须在 onNext 之前发生。如果不保留此规则，则对Subject.onNext 的并发调用将在Observer.onSubscribe 调用之前甚至同时运行，并找到可能没有准备好的消费者。

由于PublishSubject 不保留任何onNext 调用，未观察到的onNext 项目将被丢弃。根据用例，您应该改用BehaviorSubject 或yourSubject.startWith(initialValue).subscribe() 在Subject 中的任何其他onNext 之前获取值。

【讨论】：