java - 在 doOnSubscribe 中调用 subject.onNext()

Question

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

Answer 1

订阅 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 .