java - doOnNext() 中的主题未调用 Consumer.accept()

Question

我有两个 Subject，一个订阅另一个来获取更新。

Subject<Integer> subject = new Subject<>() {
    @Override
    public boolean hasObservers() {
        return false;
    }

    @Override
    public boolean hasThrowable() {
        return false;
    }

    @Override
    public boolean hasComplete() {
        return false;
    }

    @Override
    public Throwable getThrowable() {
        return null;
    }

    @Override
    protected void subscribeActual(Observer<? super InitialAPIResponse> observer) {

    }

    @Override
    public void onSubscribe(Disposable d) {

    }

    @Override
    public void onNext(Integer result) {
        Log.d(TAG, "onNext: " + apiResponse);
    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onComplete() {

    }
};

subject.doOnNext(result -> Log.d("Subject", "accept: " + result));

observableSubject
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .subscribe(subject);

observableSubject.onNext(1);
observableSubject.onComplete();

当调用onNext()时，Consumer的accept()在doOnNext()中提供> 不是。即使根据文档

Observable.doOnNext()

Modifies the source ObservableSource so that it invokes an action when it calls onNext.

Scheduler:doOnNext does not operate by default on a particular Scheduler

onNextthe action to invoke when the source ObservableSource calls onNext

return the source ObservableSource with the side-effecting behavior applied

据我从文档中了解到，可观察对象应该调用 doOnNext() 中的 Consumer。

我正在学习RxJava所以也许我在这里做错了什么......

Answer 1

有两个问题:

1.

subject.doOnNext(result -> Log.d("Subject", "accept: " + result));

在上面的代码中，doOnNext的结果没有被订阅。与许多其他运算符一样，doOnNext 本身并不订阅上游。改成这样，例如:

subject.doOnNext(result -> Log.d("Subject", "accept: " + result)).subscribe();

2.

observableSubject
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .subscribe(subject);

observableSubject.onNext(1);
observableSubject.onComplete();

在上面的代码中，onComplete 在 .onNext 之后立即调用。这可能会在发出值时导致计时问题。

将上面的代码更改为

observableSubject
            .subscribe(subject); // subscribe on the same thread so that everything happens sequentially. 

observableSubject.onNext(1);
observableSubject.onComplete();

或

Subject<Integer> observableSubject = BehaviorSubject.create();
observableSubject
            .subscribeOn(Schedulers.io())
            .observeOn(Schedulers.io())
            .subscribe(subject);

observableSubject.onNext(1);
// observableSubject.onComplete(); // don't call onComplete/