CompletableFuture不会忽略线程

Question

我的异步方法有问题。它工作正常，但线程数一直在增加。

下面是我的代码示例：

ExecutorService executorService = Executors.newFixedThreadPool(marketplaces.size());

public void createCache() {
        List<CompletableFuture<List<OrderResponseDto>>> futures = marketplaces.stream().map(
                m -> CompletableFuture.supplyAsync(m::trades, executorService)
                        .applyToEither(timeoutAfter(TIMEOUT_SECONDS, TimeUnit.SECONDS), Function.identity())
                        .exceptionally(error -> {
                            log.warn("Trades failed {}", error);
                            return Collections.emptyList();
                        })
        ).collect(toList());

    Map<TradePlatform, List<OrderResponseDto>> collect = futures.stream()
            .map(CompletableFuture::join)
            .flatMap(List::stream)
            .collect(groupingBy(OrderResponseDto::getMarketplace));
    marketplaceCache.putAll(collect); 
}

位于GitHub的完整项目：https://github.com/rublin/KarboMarketplaceExplorer

直接链接到类

下面是一个测试，涵盖了这种行为。

Answer 1

总结评论中列出的调查结果：

这是由timeoutAfter（）方法引起的线程泄漏（请参阅链接的源代码）。

此方法每次调用时都会创建一个新的线程池，其中包含一个线程，但没有任何方法可以关闭该池，从而停止线程。

由于在createCache（）中创建的每一个CompletableFuture都会调用一次，所以线程的总数只会增加。

解决方法是在应用程序的生命周期内重用线程池，并确保它们在退出时正确关闭（另一个需要考虑的问题是如何调整线程池的大小，以及使用什么类型的线程池，尤其是当提交给它们的任务可能会阻塞时）。

另见内森·休斯对该问题的评论。