RxJavarx.exceptions.MissingBackpressureException与过滤器和映射

Question

我刚开始使用RxJava/RxAndroid，在理解如何正确处理背压方面有一些问题。

我有一个文件扫描器，可以扫描目录并发出文件。应尽快处理这些文件，并且不跳过任何文件。

所以管道看起来像这样：<代码>可观察

不幸的是，我收到了rx。例外情况。缺少背压异常错误。所以我读了关于背压的书，如果我理解正确的话，无损耗选项只是缓冲区和窗口。

我试过了BackPressureBuffer（）、buffer（）和window（）。虽然所有的onBackpressureX（）命令似乎都没有效果，但buffer（）将项目分组到一个列表中

1. 我应该如何过滤这些组？过滤器（

一些反馈甚至提示会有很大帮助并受到赞赏。




Answer 1

我想我找到了问题的解决方案：此代码不起作用：

Observable<File> task = scanner.getProcessDirectoryTask(mountPoint);
Subscription _subscription = task
       .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .onBackpressureBuffer(10000)
        .filter(new Func1<File, Boolean>() {
            @Override
            public Boolean call(File file) {
                return file.getAbsolutePath().endsWith("xyz");
            }
        })
        .buffer(100)
        .subscribe(new Observer<List<File>>() { /*whatever you want to do*/ }

但是，此代码正在运行：

Observable<File> task = scanner.getProcessDirectoryTask(mountPoint);
Subscription _subscription = task
        .onBackpressureBuffer(10000)
        .filter(new Func1<File, Boolean>() {
            @Override
            public Boolean call(File file) {
                return file.getAbsolutePath().endsWith("xyz");
            }
        })
        .buffer(100)
       .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<List<File>>() { /*whatever you want to do*/ }

因此，订阅（）和观察（）的顺序似乎有很大的不同！

我的第三个问题有点离题，但仍然悬而未决。也许有人可以对此发表评论。