在计划任务中使用Flux

Question

我正在从事一个Spring Webflux项目，在计划任务中发布和使用Flux时遇到了一个问题。

@Scheduled(fixedRate = 20*1000)
fun updateNews() {
    try {
        logger.info("Automatic Update at: ${LocalDateTime.now()}")
        articleRepository.saveAll(
                sourceRepository.findAll().publishOn(Schedulers.parallel())
                        .map { source -> source.generate() }
                        .flatMap { it?.read() ?: Flux.empty() })
                        .timeout(Duration.ofSeconds(20)
        ).subscribeOn(Schedulers.parallel())
    } catch(e: Throwable) {
        logger.log(Level.SEVERE, "Error in Scheduler", e)
    }
}

我配置的调度程序：

ConcurrentTaskScheduler(Executors.newScheduledThreadPool(3))

除非我在最后故意阻止，否则这项任务永远不会完成：

.then().block()

我最初没有费心直接引用发布/订阅计划程序，我尝试了所有看似合理但没有效果的选项。

我的日志事件发生了，但当来自调度程序的该任务的线程死亡时，通量也会被丢弃；即使在我指定发布和订阅行为后，它们应该在自己的线程池中？

我想使这个行动完全被动，任何建议将不胜感激。

Answer 1

@计划没有与Flux集成，因此如果您返回它，它将不知道如何处理Flux。结合这一事实，在Reactor（和一般的反应流）中，通常不会发生任何事情，直到您订阅（），您就可以开始看到哪里出了问题。

block（）实际上是subscribe（）的一种形式，这就是为什么一旦将其添加到代码中，它就会起作用。这实际上可能是最好的选择，因为您需要将一段反应式代码（从反应式存储库）连接到命令式阻塞世界（您的@Scheduled fun）。