java - Rx Java zip 在任何主题完成时完成

Question

我正面临这个问题。据我所知，RxJava 中的 zip 方法等待所有可观察对象完成。

但是我得到了另一种行为。

这是我的代码 fragment

 private PublishSubject<Void> firstSubject;
    private PublishSubject<Void> secondSubject;


    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        loadData();
        mDrawerHeaderView.postDelayed(new Runnable() {
            @Override
            public void run() {
              //  getSecondSubject().onNext(null);
            }
        }, 1000);
        mDrawerHeaderView.postDelayed(new Runnable() {
            @Override
            public void run() {
                getFirstSubject().onCompleted();
            }
        }, 1000);
    }

    protected PublishSubject<Void> createFirstSubject() {
        firstSubject = PublishSubject.create();
        return firstSubject;
    }

    protected PublishSubject<Void> createSecondSubject() {
        secondSubject = PublishSubject.create();
        return secondSubject;
    }

    protected PublishSubject<Void> getFirstSubject() {
        return firstSubject;
    }

    protected PublishSubject<Void> getSecondSubject() {
        return secondSubject;
    }

    private void loadData() {
        Observable<Void> firstSubject = createFirstSubject();
        Observable<Void> secondSubject = createSecondSubject();
        Observable<Boolean> allDataTask = Observable.zip(firstSubject, secondSubject, new Func2<Void, Void, Boolean>() {
            @Override
            public Boolean call(Void aVoid, Void aVoid2) {
                return true;
            }
        });
        allDataTask
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Boolean>() {
                    @Override
                    public void onCompleted() {
                        Notifications.showSuccessMessage(getApplicationContext(), "COMPLETE");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Notifications.showErrorMessage(getApplicationContext(), "ERROR");

                    }

                    @Override
                    public void onNext(Boolean aBoolean) {
                        Notifications.showSuccessMessage(getApplicationContext(), "NEXT");
                    }
                });
    }

在这种情况下，我收到了 COMPLETE 消息，但我原以为不会收到任何消息，因为第二个主题尚未完成。我做错了什么？请帮助我获得所需的行为。

Answer 1

是的，它有效 as expected .在这里接收 onCompleted() 是非常有意义的，因为如果一个流完成了，只要它发出的所有元素都被“压缩”，就没有办法再“压缩”任何东西，所以它“完成”了。您还可以玩序列 here .