反应堆：延迟不适用

Question

因此，我想使用Flux来运行每个单声道任务并等待每个任务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()

，即使它在同一线程（测试工作者）上运行，也似乎无法正常工作。我在这里做错了什么？

编辑：我不太了解它的工作原理，但是我没有在Flatmap的内部添加延迟内容，而是在其顶部添加延迟元素，并且可以正常工作。

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是在Prefatch和并发256和32中，这意味着FlatMap中的代码将在Paralel中运行，并且所有项目将延迟2秒，但将执行。 （订阅）同时
如果您想将其放入其中，则需要与Prefatch 1使用ConcatMap 1，

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

这将使代码在内部等待，直到预览完成请求项目1 x 1，

但更干净的解决方案是在之后延迟它