io 反应器中带有平面图的背压答案

【问题标题】：Backpressure with flatmap in io reactorio 反应器中带有平面图的背压
【发布时间】：2018-08-24 18:20:46
【问题描述】：

我们正在使用带有响应式 spring data cassandra 模块的 springboot 2.0。我们有以下代码

Flux.range(1,10000).flatMap(this::getData).subscribe(new BaseSubscriber<List<Data>>() {
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            subscription.request(1);
        }

        @Override
        protected void hookOnNext(List<Data> value) {
            request(1);
        }
    });

public Mono<List<Data>> getData(int n){
    //cassandra reactive call to get List of data
}

我们面临的问题是 getData 被调用以获得最大整数值，即使我们使用 BaseSubscriber 有背压。由于这个请求在 cassandra 超时。我认为当我们使用 flatmap 并且 getData 返回单声道时，不应用背压。我们必须使用平面图，因为在单声道错误时，我们返回 mono.empty 以保持其余整数的请求数据不断变化。有人可以帮助在这种情况下应用背压吗？

【问题讨论】：

是getData 真的要求Integer.MAX_VALUE 吗？你可以在你的getData() 中添加一个.log() 来确认吗？

标签： spring-boot project-reactor spring-data-cassandra

【解决方案1】：

有flatMap 的重载，您可以在其中设置concurrency 参数（实际上更多是并行度因素）和prefetch 参数（默认为256）。最后一个是第一次请求内部通量的数量，因此您可以将其设置为1。

【讨论】：