正确处理来自异步调用的反压力

Question

我有一些大致看起来像这样的代码：

Completable handle(Observable<Object> inbound) {
  return inbound.buffer(1, TimeUnit.Second, 250)
    .filter(l -> !l.isEmpty())
    .flatMapToCompletable(this::sendToClient);
}

Completable sendToClient(List<Object> list) {
  // call IO asynchronously
}

入站是一个冷可观察的，它在标准的同时循环中产生（但是在另一个线程上！）。在未配置时，获取一些数据并调用onNext（）。

我面临的问题是sendToClient太“慢”，它跟不上inbound生成数据的速度。这是一个异步操作，本质上是将对内存中某个位置的客户机的调用排队（无法访问此代码），并最终填充堆。

我需要找到一种方法，当这些调用开始堆积时，让入站停止生成。有没有一种方法可以使用内置操作符来实现这一点？

Answer 1

我可能会建议创建一个线程池来处理sendToClient操作。这可以支持等待处理的请求队列，但也可以指定队列已满时的行为（即背压过大）。

看看java.util.concurrent.线程池执行器