RxJava3 Flowable无界背压相关技术问题咨询

Reactive Streams 基于背压构建的特性非常出色，我希望能更深入理解RxJava3 API的使用方式。针对GUI事件，采用丢弃溢出数据的方案完全可行，但在处理文件或Kafka主题这类场景时，通常会受限于数据库插入速度，因此需要采用合适的RxJava模式来确保不丢失数据且不拖慢生产者。

我曾在RxJava群组发布相关问题，被建议在此提问，现咨询以下问题：

1. 为何Flowable#subscribe没有指定订阅请求大小的版本，而是统一使用FlowableInternalHelper.RequestMax.INSTANCE？
2. 默认订阅大小为何与默认缓冲区大小（128）不相关？比如默认设为64是否更合理？
3. 我对API的理解是否正确：要实现不丢失数据、不报错的背压，必须使用如下示例中的subscribeWith(new DisposableSubscriber...)，且API没有对应的快捷方式？

.subscribeWith(
    new DisposableSubscriber<List<Buffer>>() {
      @Override
      public void onStart() {
        request(1);
      }

      @Override
      public void onNext(List<Buffer> buffers) {
        // ...
        request(1);
      }

      @Override
      public void onError(Throwable throwable) {
        // ...
      }

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

// 或使用vertx promise的类似示例
.subscribeWith(new RequestSub<>(promise::fail, promise::complete));

@RequiredArgsConstructor
public class RequestSub<T> extends DisposableSubscriber<T> {

  private final Consumer<? super Throwable> onError;
  private final Runnable onComplete;

  @Override
  public void onStart() {
    request(1);
  }

  @Override
  public void onNext(T next) {
    request(1);
  }

  @Override
  public void onError(Throwable throwable) {
    onError.accept(throwable);
  }

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

另外，我编写了一个rxjava-agent工具，以便更便捷地检测无界订阅的使用场景。

问题解答 #

1. 为何Flowable#subscribe没有指定请求大小的重载版本 #

RxJava3设计Flowable#subscribe的简化重载时，核心目标是提供开箱即用的便捷性，默认请求Long.MAX_VALUE（即RequestMax.INSTANCE）是为了适配大多数不需要精细背压控制的场景——比如快速搭建原型、处理无界但流速可控的数据源。如果每个subscribe都强制要求指定请求大小，会大幅增加API的复杂度，违背简化接口的设计初衷。

若需要自定义请求大小，RxJava3提供了subscribeWith(DisposableSubscriber)或subscribe(Subscriber)这类底层API，让开发者自行控制背压逻辑，这是分层设计的体现：上层简化接口满足通用需求，底层接口满足定制化需求。

2. 默认订阅大小与默认缓冲区大小（128）无关的原因 #

默认请求Long.MAX_VALUE和默认缓冲区128是两个不同维度的设计：

• 默认缓冲区128是Flowable在异步操作（如observeOn）时的缓冲上限，目的是平衡内存占用和处理效率，避免无界缓冲导致OOM。
• 默认请求Long.MAX_VALUE是订阅时向上游发起的无界请求，对应“信任上游流速可控”的场景，这和异步缓冲区的设计目标没有直接关联。

至于默认设为64是否合理，这完全取决于具体业务场景，但RxJava3选择无界请求作为默认，是因为它能覆盖更多通用场景——如果上游本身是背压友好的，无界请求不会有问题；如果上游流速不可控，开发者自然会通过onBackpressureBuffer/onBackpressureDrop或自定义Subscriber来调整，这属于开发者需要根据业务场景做出的选择，框架不会强制一个“通用合理”的数值。

3. 关于实现可靠背压的API理解 #

你的理解是对的，但有补充：

• 要实现不丢失数据、不报错的可靠背压，确实需要通过DisposableSubscriber（或自定义Subscriber）手动调用request(n)来控制上游的发送速率，确保下游处理完一批数据后再请求下一批，避免溢出。
• RxJava3确实没有直接的快捷方式来生成这种“逐批请求”的订阅者，但你可以自己封装类似RequestSub的通用订阅者，或者使用flatMap配合Flowable.just(item)这类方式间接实现背压控制（不过手动实现DisposableSubscriber是最直接高效的方式）。

另外，你编写的rxjava-agent工具很实用，能帮助开发者快速发现无界订阅的风险点，避免因默认无界请求导致的内存溢出或数据积压问题。

内容的提问来源于stack exchange，提问作者kodstark