java - 有条件地将 Mono 与 Flux 结合起来

Question

我需要合并两个响应式(Reactive)发布器(Mono 和 Flux)的结果。我尝试使用 zip 和 join 函数来完成此操作，但我无法满足两个特定条件:

1. 结果应包含与 Flux 发出的元素一样多的元素，但相应的 Mono 源应仅调用一次(仅此条件可以通过 join 实现)
2. 当 Flux 为空时，链应该完成而无需等待 Mono 元素

第一个条件的解决方案在 Combine Mono with Flux 中给出。条目(粘贴在下面)。但我无法在不阻塞链条的情况下实现第二个条件 - 我想避免这种情况。

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 为空时取消整个操作，您可以执行以下操作

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 完成但不希望 join 挂起，您可以执行以下操作

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);