在RxJava中处理背压而不丢弃项目或序列化

Question

简言之，有没有解决方案可以解决RxJava中的背压问题，而不需要删除项、序列化操作或无限缓冲？

把下面的任务当作一个有用的例子。

1. 从磁盘读取数据到内存
2. 压缩数据
3. 通过网络传输压缩数据

简单的方法是在单个后台线程上顺序执行所有任务，如：

observeBlocksOfFileContents(file).
    .subscribeOn(backgroundScheduler)
    .map(compressBlock)
    .subscribe(transmitBlock);

虽然这样做没有问题，但从性能角度来看，这是次优的，因为运行时是所有三个操作的总和，而不是并行运行时的最大值：

observeBlocksOfFileContents(file).
    .subscribeOn(diskScheduler)
    .observeOn(cpuScheduler)
    .map(compressBlock)
    .observeOn(networkScheduler)
    .subscribe(transmitBlock);

然而，如果数据从磁盘读取的速度快于压缩和传输的速度，这可能会由于背压而失败。通常的背压解决方案是不可取的，原因如下：

1. 丢弃项目：文件必须完整传输，没有丢失的片段
2. 在单线程上串行：流水线的性能提升丢失
3. 调用堆栈阻塞：RxJava不支持
4. 增加观察缓冲区：内存消耗可能会变成文件大小的几倍
5. 在不丢失Backpress的情况下重新实现观察异常：大量的工作和破坏流畅的API

还有其他解决方案吗？或者这根本不符合ReactiveX可观察模型？

Answer 1

6） 实现observeBlocksOfFileContents，以支持背压。

文件系统已经是基于拉的（InputStream.read（）发生在您想要的时候，而不是向您抛出），所以请考虑一个合理的块大小，并在每个请求中读取：

Observable.create(SyncOnSubscribe.createStateful(
    () -> new FileInputStream("file.dat")
    (in, out) -> {
        byte[] buf = new byte[4096];
        int r = in.read(buf);
        if (r < 0) {
            out.onCompleted();
        } else {
            if (r == buf.length) {
                out.onNext(buf);
            } else {
                byte[] buf2 = new byte[r];
                System.arraycopy(buf, 0, buf2, 0, r);
                out.onNext(buf2);
            }
        }

    }, 
    in -> in.close()
));

（为简洁起见，请尝试省略。）