在forEach循环java 11中应用completableFuture答案

【问题标题】：applying completableFuture inside a forEach loop java 11在forEach循环java 11中应用completableFuture
【发布时间】：2021-03-24 15:44:56
【问题描述】：

我的 Spring Boot 应用程序中有一个计划任务，我想更快地运行它。

我的任务包括获取一些数据（订阅），并为每个订阅调用 API 客户端并向我的 kafka 代理发送消息。

这是我的代码

    @Scheduled(cron="0 0 12 1/1 * ? *",zone = "Europe/Paris")
    public void execute() {

        List<StoreSubscription> storeSubscriptions = subscriptionRepository.findAllByExpirationDate(LocalDate.now().plusMonths(1));

        storeSubscriptions.forEach(subscription -> {
            CompletableFuture.supplyAsync( () -> this.message(subscription))
                    .thenApply(message -> kt.send(SUBSCRIPTION_EXPIRED_REMINDER_EARLY,message));
        });
    }

消息是异步创建的吗？是否在不等待上一条消息结束的情况下创建消息？

谢谢。

【问题讨论】：

你可以在这里添加一个日志语句来调试你的假设

标签： java spring-boot asynchronous apache-kafka completable-future

【解决方案1】：

是的，第二条消息不会等待第一条消息结束。发生这种情况是因为 CompletableFutures 没有组合。

如果你想让它们一个接一个地被执行，你需要这样的东西：

CompletableStage loop = CompletableFuture.completedFuture(null);
storeSubscriptions.forEach(subscription -> {
    loop = loop.thenCompose( unusued -> CompletableFuture
                           .supplyAsync( () -> this.message(subscription))
                           .thenAccept(message -> kt.send(SUBSCRIPTION_EXPIRED_REMINDER_EARLY,message));
});

此解决方案的缺点是它可能会在内存中创建许多对象。为避免此问题，您可以使用IBM async utils:

Iterator iterator = storeSubscriptions.iterator();
AsyncTrampoline.asyncWhile( () -> CompletableFuture
                           .supplyAsync( () -> this.message(iterator.next()))
                           .thenAccept(message -> kt.send(SUBSCRIPTION_EXPIRED_REMINDER_EARLY,message))
                           .thenApply( r -> iterator.hasNext() )
)

使用 Maven，您可以将其添加到您的项目中：

<dependency>
    <groupId>com.ibm.async</groupId>
    <artifactId>asyncutil</artifactId>
    <version>0.1.0</version>
</dependency>

这个库实现了Trampoline pattern

【讨论】：