java - 你能 Flux.zip 一个单声道和一个助焊剂，并为每个助焊剂值重复单声道值吗？

Question

是否可以执行以下代码之类的操作？我有一个进行 API 调用的服务和另一个返回值流的服务。我需要通过 API 调用返回的值来修改每个值。

return Flux.zip(
                     someMono.get(),
                     someFlux.Get(),
                     (d, t) -> {
                         //HERE D IS ALWAYS THE SAME AND T IS EVERY NEW FLUX VALUE
                     });

我已经尝试为 Mono 使用 .repeat() 并且它可以工作，但是每次有一个新的 Flux 值并且它是一个 API 调用时它都会调用该方法，所以它并不好。

可能吗？

Answer 1

这将说明如何将通量与单声道相结合，这样每次通量发射时，单声道也会被发射。

假设你有一个通量和一个像这样的单声道：

 // a flux that contains 6 elements.
 final Flux<Integer> userIds = Flux.fromIterable(List.of(1,2,3,4,5,6));

 // a mono of 1 element.
 final Mono<String> groupLabel = Mono.just("someGroupLabel");

首先，我将向您展示尝试压缩我尝试的 2 的错误方法，我认为其他人会尝试：

 // wrong way - this will only emit 1 event 
 final Flux<Tuple2<Integer, String>> wrongWayOfZippingFluxToMono = userIds
         .zipWith(groupLabel);

 // you'll see that onNext() is only called once, 
 //     emitting 1 item from the mono and first item from the flux.
 wrongWayOfZippingFluxToMono
         .log()
         .subscribe();

 // this is how to zip up the flux and mono how you'd want, 
 //     such that every time the flux emits, the mono emits. 
 final Flux<Tuple2<Integer, String>> correctWayOfZippingFluxToMono = userIds
         .flatMap(userId -> Mono.just(userId)
                 .zipWith(groupLabel));

 // you'll see that onNext() is called 6 times here, as desired. 
 correctWayOfZippingFluxToMono
         .log()
         .subscribe();

Answer 2

您可以使用运算符完成此cache操作。取消注释Fluxwithout cache，您将看到对getNum==的调用次数100。有了cache它就会1。

public class RepeatableMono {

  private static AtomicInteger numberOfCalls = new AtomicInteger(0);

  static Integer getNum() {
    System.out.println("GetNum Called: " + numberOfCalls.incrementAndGet());
    return 0;
  }

  public static void main(String[] args) {
    // This will print `GetNum Called: ` 100 times.
    //Flux<Integer> neverEndingFlux = Mono.defer(() -> Mono.just(getNum()))
    // .repeat();

    // This will print `GetNum Called: ` 1 times.
    Flux<Integer> neverEndingFlux = Mono.defer(() -> Mono.just(getNum()))
        .cache()
        .repeat();

    Flux<Integer> stream = Flux.range(1, 100);

    Flux.zip(neverEndingFlux, stream, (x, y) -> x + " " + y)
        .subscribe(System.out::println);
  }
}