Java 迭代列表<?扩展通量>

Question

我的方法得到了 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
 }

Answer 1

这似乎是一种反模式。但是有一些基本的错误。

1. map(children -> children.collectList()) 将返回 Mono<List<Child>>
2. forEach(child -> run(child); 你忘记了一个右括号，应该是forEach(child -> run(child));。
3. 但它不会编译，因为 child 将是 Mono<List<Child>> 而不是 Child
4. 当您使用响应式编程时，订阅前什么都没有发生

你真正需要做的是像

Flux.concat(childFlux).subscribe(this::run)

连接在一个 Iterable 中提供的所有源，转发由下游源发出的元素。

或者

Flux.merge(childFlux).subscribe(this::run)

将数组/可变参数中包含的 Publisher 序列中的数据合并到交错的合并序列中。与 concat 不同，源被热切地订阅。