Reactor core Flux delayElement Not Working When Duration is under 1ms

Viewed 24

all My Project use reactor kafka version 1.1.0 RELEASE and reactor-core 3.2.1. I want to consume slowly so I use delayElement, like this.

ReceiverOptions<String, String> options = receiverOptions.subscription(Collections.singleton(topic)) // kafka.consumer.upsert.topic
            .addAssignListener(partitions -> log.debug("onPartitionsAssigned {}", partitions))
            .addRevokeListener(partitions -> log.debug("onPartitionsRevoked {}", partitions));

KafkaReceiver.create(options).receive()
             .delayElements(Duration.ofNanos(800000L))
             .doOnError(this::onError)
             .subscribe(this::onMessage);

I intend to consume about 1200 data per second. However, it takes a very short time to consume data. Maybe micro sec? so I change Duration to 1 millisec. so It was worked consume about 1000 data per second.

I add My test code here.

ArrayList<Integer> data = new ArrayList<>();

        for (int i = 0; i <= 5000; i++) {
            data.add(i);
        }

        log.info("start");

        Flux.fromIterable(data)
            //.doOnNext(t -> log.info("doOnNext : {}", t.toString()))
            .delayElements(Duration.ofNanos(99999))
            .subscribe(n -> {
                //log.info("data : {}", n);

                if (n == 5000) {
                    log.info("finish");
                }
            });
//duration is 0.99ms  It take very short time.
2022-09-21T12:50:04.961+09:00 INFO --- [           main] c.g : start
2022-09-21T12:50:05.104+09:00 INFO --- [     parallel-9] c.g : finish
//duration is 1ms It take very long time.
2022-09-21T12:52:52.048+09:00 INFO --- [           main] c.g : start
2022-09-21T12:54:08.264+09:00 INFO --- [     parallel-9] c.g : finish

I searched for the above phenomenon, but I couldn't find it. Has anyone experienced the same phenomenon as me?

plz help me. Thank you.

0 Answers
Related