没有使用 webFlux 和 netty 调用 ReactiveCosmosRepository答案

【问题标题】：ReactiveCosmosRepository is not being invoked with webFlux and netty没有使用 webFlux 和 netty 调用 ReactiveCosmosRepository
【发布时间】：2022-01-15 04:46:12
【问题描述】：

我有以下反应式存储库：

@Repository
public interface FooCosmosRepository extends ReactiveCosmosRepository<Foo, String> {
}

我使用它如下：

    @Override
    public Mono<FooResponse> getFooDetails() {

        FooResponse fooResponse = new FooResponse();
        fooResponse.setCount(1000);

        List<Foo> fooList = new ArrayList<>();

        repository.findAll().collectList().flatMap(e ->{

       //This is not invoked. findAll return Flux<T> in this case Flux<Foo>
            for (Foo foo : e) {
                fooList.add(foo);
            }
            return null;
        });

        fooResponse.setFooList(fooList);

        return Mono.just(fooResponse);
    }

FooResponse 定义如下：

@NoArgsConstructor
@Data
@FieldDefaults(level = AccessLevel.PRIVATE)
public class FooResponse {
    int rowCount;
    List<Foo> fooList;
}

我无法阻止，因为我收到错误消息 Iterating over a toIterable() / toStream() is blocking, which is not supported in thread reactor-http-nio-6

我也无法从该方法返回Flux<T>。我需要返回Mono<FooResponse>。如何查询存储库，实际获取/收集响应并添加到列表中？

有什么想法吗？

【问题讨论】：

标签： spring spring-boot spring-data azure-cosmosdb spring-webflux

【解决方案1】：

这是因为你是在命令式而不是反应式地编码。并且您正在破坏链条，这意味着 Reactor 无法完成组装阶段，然后在订阅阶段执行。

@Override
public Mono<FooResponse> getFooDetails() {
    return repository.findAll()
        .collectList()
        .map(list -> {
            FooResponse fooResponse = new FooResponse();
            fooResponse.setCount(1000);
            fooResponse.setList(list);
            return fooResponse;
        });
    }

这是基本反应堆，我推荐以下链接：

Reactor Core Features

Flight of the flux

【讨论】：