实现轮询第 3 方 API 并更新数据库的计划作业的有效方法

Question

我有一个 Spring boot 应用程序，它有一个每 1.5 秒运行一次的计划作业。其目标是从第 3 方 API 获取数据，用结果更新数据库（如果需要）并重复。在之前的更新完成之前，不应开始下一个 api 调用。

为了简单起见，我们假设代码如下所示

@Scheduled(initialDelay = 3000, fixedDelay = 1500)
public void loadUpdates() {
    List<Item> recentItems = apiClient.getUpdatesAfter(lastUpdateAt);

    List<CompletableFuture<Void>> tasks = new ArrayList<>();
    for(Item item: recentItems) {
        tasks.add(CompletableFuture.runAsync(
                () -> handleItemUpdates(item),
                itemUpdateExecutor
        ));
    }

    // need to wait for all updates to finish
    CompletableFuture.allOf(tasks.toArray(CompletableFuture[]::new)).join();
    
    doSomethingElseHere();
    
    saveTheLastUpdateTime();
}

一开始效果很好，但现在随着数据量的增长，这种方法需要很多时间。

让我们假设处理代码工作正常并且数据库查询是最佳的

问题是如何才能以更好的方式做到这一点？

我想过用队列（rabbitmq）替换线程并使用多个实例进行处理，但是如何确保在所有作业完成之前不会开始下一次迭代？

欢迎任何建议 - Spring Integration、Apache Camel 或任何其他解决方案/框架/库/队列/等。

提前谢谢您。

Answer 1

“让我们假设......数据库查询是最佳的。”

我认为这个假设可能是你的问题。通过批量更新数据库，您将获得更好的性能。

要考虑的另一件事是

CompletableFuture.runAsync

在公共 fork-join 池中运行任务。但据我所知，fork-join 池不能很好地处理执行阻塞 I/O 的任务；见

• Java 的 fork-and-join 线程池适合执行 IO 密集型任务吗？
• Java ForkJoinPool 中阻塞 IO 期间线程的行为

等等。

IMO，您最好使用传统线程池而不是分叉连接池创建自己的

ExecutorService

。事实上，如果您正在为 Java 21 及更高版本进行编码，则可以使用

newVirtualThreadPerTaskExecutor

(javadoc]。这为您提供了一个执行器服务，其中包含使用虚拟线程的无界线程池。

无论您使用哪种线程池，

invokeAll

方法都会提交所有任务并等待它们完成。