我的方法获取 Flux。 如何迭代 Flux？ 我想检查它的对象并对每个 child 进行操作。
public void write(List<? extends Flux<Child>> childFlux) throws Exception {
childFlux.stream()
.map(children -> children.collectList())
.forEach(child -> run(child); //not compile
}
public void run(Child child) {
//TO DO
}
最佳答案
这似乎是一种反模式。然而，存在一些基本错误。
-
map(children -> children.collectList())将返回Mono<List<Child>> -
forEach(child -> run(child);你忘记了一个右括号，应该是forEach(child -> run(child));. - 但它不会编译，因为 child 将是
Mono<List<Child>>而不是Child - 当您使用响应式编程时，订阅之前不会发生任何事情
你真正需要做的事情是这样的
Flux.concat(childFlux).subscribe(this::run)
Concatenate all sources provided in an Iterable, forwarding elements emitted by the sources downstream.
或者
Flux.merge(childFlux).subscribe(this::run)
Merge data from Publisher sequences contained in an array / vararg into an interleaved merged sequence. Unlike concat, sources are subscribed to eagerly.
关于Java 迭代列表 <?扩展通量>，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/57998700/