为什么在使用BackPreasureBuffer时会出现MissingBackpreasureException？

Question

我想在RxJava中实现一个下载一些文件的处理队列。我想下载的文件数量可能高达100个左右。

一切都是在Android上使用RxJava 1.1.1开发的

PublishSubject<URL> publishSubject = PublishSubject.create();
_subject = new SerializedSubject<>(publishSubject);

_subscription = _subject
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .subscribe(_getObserver());  // Observer

_subject.onBackpressureBuffer(10000, new Action0() {
    @Override
    public void call() {
        Log.d(TAG, "onBackpressureBuffer called");
        return;
    }
});

// download a file
_subject.onNext(aValidURL);

我做错了什么？

Answer 1

在RxJava中，应用运算符时，会得到一个新的可观察实例，该实例具有修改后的行为，但原始实例保持不变。在这里，您对_subject调用了onbackpressurebuffer，但不使用它的结果，否则调用不执行任何操作。您需要按顺序应用它：

PublishSubject<URL> publishSubject = PublishSubject.create();
_subject = new SerializedSubject<>(publishSubject);

_subscription = _subject
                .onBackpressureBuffer(10000, new Action0() {
                    @Override
                    public void call() {
                        Log.d(TAG, "onBackpressureBuffer called");
                        return;
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .subscribe(_getObserver());  // Observer

// download a file
_subject.onNext(aValidURL);