有条件地将Mono与Flux结合

Question

我需要结合两个反应性出版商的结果——Mono和Flux。我尝试使用zip和join函数来实现这一点，但我无法满足两个特定条件：

1. 结果包含的元素应与通量发射的元素一样多，但相应的Mono源只应调用一次（仅此条件可通过join实现）
2. 当通量为空时，链应在不等待单一元素的情况下完成

第一个条件的解决方案出现在结合单声道和通量条目中（粘贴在下面）。但是我无法在不阻塞链的情况下实现第二个条件——这是我想避免的。

Flux<Integer> flux = Flux.concat(Mono.just(1).delayElement(Duration.ofMillis(100)),
        Mono.just(2).delayElement(Duration.ofMillis(500))).log();

Mono<String> mono = Mono.just("a").delayElement(Duration.ofMillis(50)).log();

List<String> list = flux.join(mono, (v1) -> Flux.never(), (v2) -> Flux.never(), (x, y) -> {
    return x + y;
}).collectList().block();

System.out.println(list);

Answer 1

如果通量为空，则要取消整个操作，可以执行以下操作

Flux<Integer> flux = Flux.concat(Mono.just(1).delayElement(Duration.ofMillis(100)),
            Mono.just(2).delayElement(Duration.ofMillis(500))).log();

//Uncomment below and comment out above statement for empty flux

//Flux<Integer> flux = Flux.empty();

Mono<String> mono = Mono.just("a").delayElement(Duration.ofMillis(5000)).log();

//Throw exception if flux is empty
flux = flux.switchIfEmpty(Flux.error(IllegalStateException::new));

List<String> list = flux
        .join(mono, s -> Flux.never() , s -> Flux.never(), (x, y) -> x + y)
        //Catch exception and return nothing
        .onErrorResume(s -> Flux.empty())
        .collectList().block();

    System.out.println(list);

如果你想让Mono完成，但又不想让连接挂起，你可以做如下操作

DirectProcessor<Integer> processor = DirectProcessor.create();
//Could omit sink, and use processor::onComplete in place of sink::complete
//But typically recommended as provides better thread safety
FluxSink<Integer> sink = processor.serialize().sink();

Flux<Integer> flux = Flux.concat(Mono.just(1).delayElement(Duration.ofMillis(100)),
    Mono.just(2).delayElement(Duration.ofMillis(500))).log();

//Uncomment below and comment out above statement for empty flux

//Flux<Integer> flux = Flux.empty();

Mono<String> mono = Mono.just("a").delayElement(Duration.ofMillis(5000)).log();

List<String> list = flux
     .doOnComplete(sink::complete)
     .join(mono, s -> processor , s -> processor, (x, y) -> x + y).collectList().block();

System.out.println(list);