Reactor：延迟元件不能与通量一起工作

Question

所以我想使用Flux来运行每个Mono任务，并等待每个任务2秒钟。这是我试过的

val test = { x: Int ->
      Mono.just(x)
        .log()
        .doOnNext { println("Number: $it") }
    }


Flux.fromIterable(listOf(1, 2, 3))
    .flatMap { number ->
      test(number)
         .delayElement(Duration.ofSeconds(2))
    }
    .collectList()
    //.subscribeOn(Schedulers.single()) <- just try to run on the same thread didn't work either
   .block()

结果如下

01:58:21.398 [Test worker] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
01:58:21.470 [Test worker] INFO reactor.Mono.Just.1 - | onSubscribe([Synchronous Fuseable] Operators.ScalarSubscription)
01:58:21.472 [Test worker] INFO reactor.Mono.Just.1 - | request(unbounded)
01:58:21.473 [Test worker] INFO reactor.Mono.Just.1 - | onNext(1)
Number: 1
01:58:21.477 [Test worker] INFO reactor.Mono.Just.1 - | onComplete()
01:58:21.477 [Test worker] INFO reactor.Mono.Just.2 - | onSubscribe([Synchronous Fuseable] Operators.ScalarSubscription)
01:58:21.477 [Test worker] INFO reactor.Mono.Just.2 - | request(unbounded)
01:58:21.478 [Test worker] INFO reactor.Mono.Just.2 - | onNext(2)
Number: 2
01:58:21.478 [Test worker] INFO reactor.Mono.Just.2 - | onComplete()
01:58:21.478 [Test worker] INFO reactor.Mono.Just.3 - | onSubscribe([Synchronous Fuseable] Operators.ScalarSubscription)
01:58:21.479 [Test worker] INFO reactor.Mono.Just.3 - | request(unbounded)
01:58:21.479 [Test worker] INFO reactor.Mono.Just.3 - | onNext(3)
Number: 3
01:58:21.479 [Test worker] INFO reactor.Mono.Just.3 - | onComplete()

正如您在时间戳delayElement上看到的，即使它在同一个线程上运行（测试工作人员），它似乎也不起作用。我在这里做错了什么？

编辑：我真的不明白它是如何工作的，但是我没有在平面图中添加delayElement，而是在它上面添加delayElements，它工作正常。

Flux.fromIterable(listOf(1, 2, 3))
    .delayElements(Duration.ofSeconds(2))
    .flatMap { number ->
      test(number)
    }
    .collectList()
    //.subscribeOn(Schedulers.single()) <- just try to run on the same thread didn't work either
   .block()

结果

03:21:29.825 [Test worker] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
03:21:29.926 [Test worker] INFO reactor.Flux.Array.1 - | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription)
03:21:29.929 [Test worker] INFO reactor.Flux.Array.1 - | request(32)
03:21:29.930 [Test worker] INFO reactor.Flux.Array.1 - | onNext(1)
03:21:29.935 [Test worker] INFO reactor.Flux.Array.1 - | onNext(2)
03:21:29.936 [Test worker] INFO reactor.Flux.Array.1 - | onNext(3)
03:21:29.936 [Test worker] INFO reactor.Flux.Array.1 - | onComplete()
Thread[parallel-1,5,main] What is this 1 31
Thread[parallel-2,5,main] What is this 2 33
Thread[parallel-3,5,main] What is this 3 35

Answer 1

问题是，在第一个例子中，您在flatmap中执行延迟，知道flatmap具有预批和并发256和32，这意味着flatmap中的代码将并行运行，所有项将延迟2秒，但将同时执行（订阅），如果您想将其放入，则需要将concatMap与预批1一起使用

Flux.fromIterable(listOf(1, 2, 3))
.concatmap ( number ->
  test(number)
     .delayElement(Duration.ofSeconds(2))
,1)
.collectList()
.block()

这将使内部代码等待预览完成对项目的逐个请求

但更清洁的解决方案是在