如何操作来自流量的对象，而值来自以非阻塞方式发出单个的方法？

Question

我试图用从Mono接收的数据来操作从Flux接收的对象，其中发出对象的Flux和项的Mono的方法都是不同的API调用。问题是，我无法控制线程，从Mono接收的项永远不会分配给我的对象，除非我故意阻止（）那个线程。请建议是否有任何非阻塞的方式可能为这种情况。

我还研究了调度器、订阅器、发布器，但无法找出管道。

public Flux<Object> test {

 method1().map(obj -> {
        if (obj.getTotalItems() > 20) {
            obj.setItems(method2(obj).block());
        }
        return obj;
  });
}

这里，method1是发射从API命中接收的对象的流量。

并且method2正在发出从另一个API命中中获取的项列表。

我怎样才能使这整个流动不阻塞？

Answer 1

使用flatmap运算符，可以在非阻塞公共中扁平化子流

Flux<Object> test {

 method1().flatMap(obj -> {
        if (obj.getTotalItems() > 20) {
            return method2(obj)
                     .map(result -> {
                        obj.setItems(result);
                        return obj;
                     });
        }
        return Mono.just(obj);
  });
}

flatMap允许您一次扁平化几个流，因此在长时间运行的操作中，您可以使用更高效的流程元素。

flatMap的一个缺点是它不保留元素的顺序，因此如果您有一个上游元素序列，如[1,2,3,4]和flatMap,则由于子流的异步性质，顺序可能会改变。

要保留顺序，可以使用concatmap，它一次只对流进行一次扁平化，因此可以保证将保留扁平化元素的顺序：

Flux<Object> test {

 method1().concatMap(obj -> {
        if (obj.getTotalItems() > 20) {
            return method2(obj)
                     .map(result -> {
                        obj.setItems(result);
                        return obj;
                     });
        }
        return Mono.just(obj);
  });
}

这样的对象突变方式不是最好的主意，我更喜欢在反应性编程中使用不可变的对象模式object