代码之家 › 专栏 › 技术社区 › Jordi

Java CompletableFuture:CompletableFuture协调

completable-future multithreading java

Jordi · 技术社区 · 5 年前

int pageNumber = 0;
int pageSize = 100;
ResultSetType resultSet;
do {
    resultSet = this.serveiTerritorialClientRepository
        .getOid("2.16.724.4.400", pageNumber, pageSize);
            
    // do something with resultSet
            
    pageNumber++;
} while (resultSet.getResultCount() < pageSize)

resultSet resultSet.getTotalCount() .

我想使用 CompletableFuture s

我知道如何建造 完全未来 :

CompletableFuture<ResultSetType> completableFutureOfResultSetType = 
    CompletableFuture.supplyAsync(() -> this.serveiTerritorialClientRepository
        .getOid("2.16.724.4.400", pageNumber, pageSize)
);

问题是如何协调它们。我是说,有多少 完全未来 我需要创造什么?

0 回复 | 直到 5 年前

Eugene 5 年前

这很简单:

 List<CompletableFuture<ResultSetType>> futures = new ArrayList<>();
 while (resultSet.getResultCount() < pageSize) {
    CompletableFuture<ResultSetType> completableFutureOfResultSetType = 
        CompletableFuture.supplyAsync(
            () -> this.serveiTerritorialClientRepository.getOid("2.16.724.4.400", pageNumber, pageSize)
        );
    futures.add(completableFutureOfResultSetType);       
    pageNumber++;
 }

 CompletableFuture<Void> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture<ResultSetType>[0]));

问题是 all CompletableFuture<Void> ,那么您就需要 join 他们每个人都想得到 resultSet :

futures.stream().map(CompletableFuture::join).collect(Collectors.toList());

上面的结果是您的结果集列表。即使你打电话也要小心 参加 CompletableFuture::allOf

然后,另一个可能出现的问题是,你所提供的一些期货 allOff 可能会失败。因此,您需要筛选失败的,然后重试?这取决于你的使用情况。如果要在至少一个失败的情况下失败,则需要检查: all.isCompletedExceptionally . 文件 allOf 提到如果alt至少有一个失败, 全部的完全灵巧 将报告也失败。

至于你需要多少功能,这取决于你。

在您提供的示例中,没有为定义池 supplyAsync ,因此,您可能会涉及很多线程。这 Q&A 解释了原因。通常,强烈建议为您的操作提供一个池( 供应同步 具有接受此类参数的重载方法)。在这种情况下,创建多少未来都无关紧要,因为池线程将负责调度它们。如果不进行测量,就不可能知道你需要多少。 (我们已经观察到,在我们的使用中),几个线程(2-4)就足够了。