通量到列表而不阻塞

Question

正在寻找将Flux转换为＜code＞列表

Flux.from(Collection.find())

使用反应式编程，但graph ql期望List

带Block（）的代码

public List<Test> findAll() {
        return Flux.from(testCollection.find()).collectList().block();

}

错误:-

block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-kqueue-7

在这里，我需要返回列表




Answer 1

以下示例是转换助焊剂

public static void main(String[] args) {
    Flux<String> flux = Flux.just("test1", "test2", "test3");
    List<String> list = new ArrayList<>();
    flux.collectList().subscribe(list::addAll);
    list.forEach(System.out::println);
}

Answer 2

匿名用户

我假设< code>Collection类来自某个reactor库，正在进行一些tcp/http调用。< code>collection.find返回一个< code>Flux/Mono/Publisher我假设？这不是因为< code>collectList不允许您这样做，而是因为您试图在一个< code >非阻塞线程上运行< code>block，我假设< code>collection.find在一个线程上发布元素，该线程是< code >非阻塞的一个实例，其名称为< code > reactor-http-k queue-7 ，可能是一个netty线程。

你可以看看阻止单曲订阅者.blockingGet方法，它告诉你为什么

    final T blockingGet() {
        if (Schedulers.isInNonBlockingThread()) {
            throw new IllegalStateException("block()/blockFirst()/blockLast() are blocking, which is not supported in thread " + Thread.currentThread().getName());
        }
...

如果您必须从调用方线程（调用 Flux.from）获取结果，那么您可以执行

Flux.from(testCondition.find())
.publishOn(Schedulers.boundedElastic())
.collectList()
.block()

Answer 3

正如评论中所说，你不能。反应模式是保持在一个流中。

所以，

Mono<GraphqlResponse> = Flux.just("A", "B" "C")
  .collectList()
  .map(this::someMethod);

GraphqlResponse someMethod(List<String> abcs) {
    return graphQl.doSomething(abcs);
}