为什么此通量不在单线程中执行？

Question

我的函数是这样执行的：

@EventListener(classes = {ApplicationReadyEvent.class})
public void executeSendingNotificationToServer() {
serverNotificationService.trySendNotification(msgCount, msgTime)
.delaySubscription(Duration.ofMillis(notificationServerProperties.getExecutorDelay()))
.repeat()
.subscribeOn(Schedulers.single())
.subscribe();
}

方法trySendNotification以某种方式执行(这无关紧要(。

为什么它不在单线程中执行？我显式地设置了Schedulers.single((并且文档声明它将在一个线程中执行。

相反，我可以观察到创建了多个线程(我将线程名称记录在方法中，它打印不同的名称(

Answer 1

您应该在延迟之前移动subscribeOn。

这里有一个例子，使用您的代码，但打印线程号：

Mono.fromCallable(() -> {
System.out.println("Thread = " + Thread.currentThread().getId());
return "hello world";
})
.delaySubscription(Duration.ofMillis(500))
.repeat()
.subscribeOn(Schedulers.single())
.subscribe();

正如您所说，输出显示它在不同的线程中执行：

Thread = 14
Thread = 15
Thread = 16
Thread = 17
Thread = 18
Thread = 19
Thread = 20
Thread = 21
Thread = 14

现在，如果我以前这样移动它：

Mono.fromCallable(() -> {
System.out.println("Thread = " + Thread.currentThread().getId());
return "hello world";
})
.subscribeOn(Schedulers.single()) // <- BEFORE
.delaySubscription(Duration.ofMillis(500))
.repeat()
.subscribe();

输出变为：

Thread = 14
Thread = 14
Thread = 14
Thread = 14
Thread = 14
Thread = 14
Thread = 14
Thread = 14
Thread = 14