将Mono与发射的每个通量元素组合

Question

我有一个流量和单声道如下：

Mono<MyRequest> req = request.bodyToMono(MyRequest.class);
Mono<List<String>> mono1 = req.map(r -> r.getList());;
Flux<Long> flux1 = req.map(r -> r.getVals()) // getVals() return list of Long
        .flatMapMany(Flux::fromIterable);

flux1.flatMap(id -> process(id, mono1)) 

但是传递和处理相同的mono1会导致错误只允许一个连接接收订阅服务器。我怎样才能做到以上？谢了！

Answer 1

因为这两个信息来自同一个来源，所以您可以像这样使用一个管道运行整个过程，并将这两个元素包装在元组中，或者更好，一个具有更多含义的域对象中：

Mono<MyRequest> req = // ...
Flux<Tuple2<Long, List<String>>> tuples = req.flatMapMany(r ->
        Flux.fromIterable(r.getVals())
                .map(id -> Tuples.of(id, r.getList()))
);
// once there, you can map that with your process method like
tuples.map(tup -> process(tup.getT1(), tup.getT2());

注意，这看起来很不寻常，这基本上来自于你接收到的对象的结构。