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

等待列表中的每个未来<未来>完成[关闭]

future java-8 concurrency java

markvgti · 技术社区 · 8 年前

我调用一个返回 Future ,每个元素一次 List<Principal> ,所以我最终得到了一个 List<Future<UserRecord>> .

返回的方法 将来 是库代码,我无法控制代码的运行方式,我所拥有的只是 将来 .

我想等待所有的 将来 在继续之前完成(成功或失败)。

有没有比这更好的方法:

List<Principal> users = new ArrayList<>();
// Fill users
List<Future<UserRecord>> futures = getAllTheFutures(users);
List<UserRecord> results = new ArrayList<>(futures.size());
boolean[] taskCompleted = new boolean[futures.size()];
for (int j = 0; j < taskCompleted.length; j++) {
    taskCompleted[j] = false;
}
do {
    for (int i = 0; i < futures.size(); i++) {
        if (!taskCompleted[i]) {
            try {
                results.add(i, futures.get(i).get(20, TimeUnit.MILLISECONDS));
                taskCompleted[i] = true;
            } catch (TimeoutException e) {
                // Do nothing
            } catch (InterruptedException | ExecutionException e) {
                // Handle appropriately, then...
                taskCompleted[i] = true;
            }
        }
    }
} while (allNotCompleted(taskCompleted));

对于好奇的人:

private boolean allNotCompleted(boolean[] completed) {
    for (boolean b : completed) {
        if (!b)
            return true;
    }
    return false;
}

与以下答案不同 Waiting on a list of Future 我无法控制创建 将来 s

2 回复 | 直到 8 年前

jcsahnwaldt Reinstate Monica 6 年前

您的代码可以简化很多 . 除非您在问题中没有指定要求,否则可以按以下方式编写等效版本。

List<Principal> users = // fill users
List<Future<UserRecord>> futures = getAllTheFutures(users);
List<UserRecord> results = new ArrayList<>();

for (int i = 0; i < futures.size(); i++) {
        try {
            results.add(futures.get(i).get());
        } catch (InterruptedException | ExecutionException e) {
            // Handle appropriately, results.add(null) or just leave it out
        }
    }
}

Elysiumplain 8 年前

你可以简单地做一个简化列表;从列表中删除成功的响应,并迭代直到为空。

List<Principal> users = // fill users
List<Future<UserRecord>> futures = getAllTheFutures(users);
List<UserRecord> results = new ArrayList<>();

for (int i = 0; i < futures.size(); i++) {
        try {
            results.add(futures.get(i).get(<how long you want before your application throws exception>));

        }
        catch (InterruptedException | ExecutionException e) {
            // Handle appropriately, results.add(null) or just leave it out
        }
        catch (TimeoutException timeoutEx) {
            // If the Future retrieval timed out you can handle here
        }

    }
}

由于您的目的是在继续之前收集“一组”作业,因此在本例中,等待线程索引X的返回将产生(大约)最后一个返回线程的时间成本。

或者,如果计划在任何线程失败时中止集中的所有线程,则可以使用Java 8 CompletableFuture

CompletableFuture[] cfs = futures.toArray(new CompletableFuture[futures.size()]);

    return CompletableFuture.allOf(cfs)
            .thenApply(() -> futures.stream()
                                    .map(CompletableFuture::join)
                                    .collect(Collectors.toList())
            );

Kayaman简化了代码库。