Rx-Java2 Floawable.rebatch请求不会重新批处理请求

Question

考虑以下用户情况：

Iterable<Object> it = getATerabyteOfDataOnDemand();
Flowable.fromIterable(it)
    .blockingSubscribe(v -> consumeSlowly(v));

当我执行之前的代码时，预期的结果是我应该得到一个内存不足的异常，因为它是可流动的。阻止长时间订阅请求。最大值项，迭代器将耗尽1 TB的数据，订阅者无法跟上。

为了解决这个问题，我将rebatch请求添加到我的代码中：

Iterable<Object> it = getATerabyteOfDataOnDemand();
Flowable.fromIterable(it)
    .rebatchRequests(128)
    .blockingSubscribe(v -> consumeSlowly(v));

当我执行以下代码时，我的预期结果是一切都工作得很好，正如Flowable.rebatch请求的留档所说：

该运算符允许阻止下游通过请求（Long.MAX_值）触发无界模式，或补偿小请求和频繁请求的每项开销。

在实践中，我得到了另一个内存不足的异常。如何使用背压来防止可流动的迭代器立即耗尽？

Answer 1

问题是，您有一个同步流，并且Block ingSubcribe将无法消耗的项目排队，因为fromIterable仍然在同一线程上发射。

要么你不需要<代码> Bug CuxBube < /Cube >，要么你应该考虑使用<代码> Bug KealIdable（）/Case>，每个都只在当前线程实际消耗项目时请求更多。否则，您必须使用阻塞订阅（订户）重载并手动请求：

source.blockingSubscribe(new Subscriber<T>() {
    Subscription upstream;
    @Override public void onSubscribe(Subscription s) {
        upstream = s;
        s.request(1);
    }

    @Override public void onNext(T item) {
        consumeSlowly(v);
        upstream.request(1);
    }

    @Override public void onError(Throwable ex) {
        ex.printStackTrace();
    }

    @Override public void onComplete() {
    }
});

但是请注意，您已经有了一个Iterable，您可以在不需要RxJava参与的情况下为每个应用同步使用它。