Algunos locks

for (int index = 0; index < totalDocCountToProcess; index++) {
    Message<MessageDto> messageToSend = MessageBuilder.withPayload(messageDto)
            .setHeader("timestamp_ms", System.currentTimeMillis())
            .build();
    this.streamBridge.send(PERFORMANCE_QUEUE, messageToSend);
    COUNTER.incrementAndGet();
    try {
        Thread.sleep(TimeUnit.NANOSECONDS.toMillis(globalDelayPerMessage)); (1)
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}

@Configuration
public class SpringAsyncConfig {

    private static final AtomicInteger THREAD_COUNTER = new AtomicInteger();

    @Bean
    public ScheduledExecutorService scheduledExecutorService(AppConfiguration configuration) {

        final ThreadFactory threadFactory = runnable -> {
            final Thread thread = new Thread(runnable);
            thread.setName("RabbitProducerExecutor-" + THREAD_COUNTER.incrementAndGet());
            return thread;
        };

        return Executors.newScheduledThreadPool(configuration.getCorePoolSize(), threadFactory);
    }

}

...
final CountDownLatch countDownLatch = new CountDownLatch((int) totalMessages);
for (int index = 0; index < numThreads; index++) {
    // Si hay resto, algunos threads envían un mensaje extra
    final long messagesForThisThread = messagesPerThread + (index < remainder ? 1 : 0);
    scheduledExecutorService.schedule(() -> {
        this.sendMessage(globalDelayPerMessage, messagesForThisThread, message, countDownLatch);
    }, 0, TimeUnit.MICROSECONDS); (1)
}
try {
    countDownLatch.await();
} catch (InterruptedException ex) {
    throw new RuntimeException(ex);
}
log.info("Envío completado. Tiempo total: {} Total docCount: {}", responseTimeService.formatResponseTime(), COUNTER.get());

...
for (int index = 0; index < totalDocCountToProcess; index++) {
    this.scheduledExecutorService.schedule(() -> {
        Message<MessageDto> messageToSend = MessageBuilder.withPayload(messageDto)
                .setHeader("timestamp_ms", System.currentTimeMillis())
                .build();
        this.streamBridge.send(PERFORMANCE_QUEUE, messageToSend);
        COUNTER.incrementAndGet();
        countDownLatch.countDown();
    }, globalDelayPerMessage * index, TimeUnit.NANOSECONDS); (2)
}

public Scheduler scheduler() {
    final AtomicInteger atomicInteger = new AtomicInteger(0);
    final ThreadFactory threadFactory = runnable -> {
        final Thread thread = new Thread(runnable);
        thread.setName("RabbitProducerMessage-" + atomicInteger.incrementAndGet()); (1)
        return thread;
    };
    return Schedulers.fromExecutor(Executors.newFixedThreadPool(3, threadFactory));
}

@Test
@SneakyThrows
@DisplayName("Emitimos 30 items/mensajes en total")
void reactiveMessageSenderStepVerifier() {

    final var counter = new AtomicInteger(0);

    StepVerifier.create(Flux.range(0, 3)
            .publishOn(this.scheduler())
            .flatMap(mapper -> Flux.range(0, 10)
                    .delayElements(Duration.ofMillis(33))
                    .publishOn(this.scheduler())
                    .doOnNext(onNext -> {
                        counter.incrementAndGet();
                        log.info("Send message {}", "ABC");
                    })
            ))
            .expectNextCount(30) (1)
            .verifyComplete();

}

@Configuration
public class SpringAsyncConfig {

    @Bean
    public Scheduler scheduler(AppConfiguration configuration) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(configuration.getCorePoolSize());  // Minimum number of threads in the pool
        executor.setThreadNamePrefix("RabbitProducerTaskExecutor-");  // Prefix for thread names
        executor.initialize();  // Initializes the thread pool
        return Schedulers.fromExecutor(executor);
    }

}

Flux.range(0, numThreads)
        .flatMap(index -> {
            final long messagesForThisThread = messagesPerThread + (index < remainder ? 1 : 0);
            return this.sendMessage(globalDelayPerMessage, messagesForThisThread, message, countDownLatch);
        })
        .subscribe();
try {
    countDownLatch.await();
} catch (InterruptedException ex) {
    throw new RuntimeException(ex);
}
log.info("Envío completado. Tiempo total: {} Total docCount: {}", responseTimeService.formatResponseTime(), COUNTER.get());

private Mono<Void> sendMessage(final long globalDelayPerMessage, final long totalDocCountToProcess,
                                  String messagePayload, final CountDownLatch countDownLatch) {
    log.info("Iniciando envío de {} mensajes con un delay de {} ms", totalDocCountToProcess, TimeUnit.NANOSECONDS.toMillis(globalDelayPerMessa
    MessageDto messageDto = new MessageDto();
    messageDto.setMessage(messagePayload);
    return Flux.range(0, (int) totalDocCountToProcess)
            .delayElements(Duration.ofMillis(globalDelayPerMessage))
            .publishOn(this.scheduler) (1)
            .doOnNext(this.getTimestampMs(messageDto))
            .then()
            .doOnTerminate(countDownLatch::countDown);
}

private Consumer<Integer> getTimestampMs(MessageDto messageDto) {
    return onNext -> {
        Message<MessageDto> messageToSend = MessageBuilder.withPayload(messageDto)
                .setHeader("timestamp_ms", System.currentTimeMillis())
                .build();
        this.streamBridge.send(PERFORMANCE_QUEUE, messageToSend);
        COUNTER.incrementAndGet();
    };
}