【发布时间】:2016-10-28 11:09:25
【问题描述】:
我有以下组件:
private JobInfo aggregateJobInfo() {
final JobsResult jobsResult = restClient().getJobs();
final List<String> jobIds = extractJobIds(jobsResult);
//fetch details, exceptions and config for each job
final List<JobDetails> jobDetails = jobIds.stream().map(jobId -> {
final JobDetailResult jobDetailResult = restClient().getJobDetails(jobId);
final JobExceptionsResult jobExceptionsResult = restClient().getJobExceptions(jobId);
final JobConfigResult jobConfigResult = restClient().getJobConfig(jobId);
return new JobDetails(jobDetailResult, jobExceptionsResult, jobConfigResult);
}).collect(Collectors.toList());
return new JobInfo(jobsResult, jobDetails);
}
private static List<String> extractJobIds(final JobsResult jobsResult) {
final ArrayList<String> jobIds = new ArrayList<>();
jobIds.addAll(jobsResult.getRunning());
jobIds.addAll(jobsResult.getFinished());
jobIds.addAll(jobsResult.getCanceled());
jobIds.addAll(jobsResult.getFailed());
return jobIds;
}
它只是调用一些 ENDPOINTS 并聚合一些数据。现在我正在尝试通过使用 CompletableFutures 来实现非阻塞,我之前并没有真正使用过..
private CompletableFuture<JobInfo> aggregateJobInfo() {
final CompletableFuture<JobsResult> jobsResultFuture = restClient().getJobs();
final CompletableFuture<List<String>> jobIdsFuture = jobsResultFuture.thenApply(JobInfoCollector::extractJobIds);
//fetch details, exceptions and config for each job
final CompletableFuture<List<CompletableFuture<JobDetails>>> jobDetailsFuture = jobIdsFuture.thenApply(jobIds -> {
return jobIds.stream().map(jobId -> {
final CompletableFuture<JobDetailResult> jobDetailsResultFuture = restClient().getJobDetails(jobId);
final CompletableFuture<JobExceptionsResult> jobExceptionsFuture = restClient().getJobExceptions(jobId);
final CompletableFuture<JobConfigResult> jobConfigFuture = restClient().getJobConfig(jobId);
return jobDetailsResultFuture.thenCompose(jobDetailResult -> {
return jobExceptionsFuture.thenCombine(jobConfigFuture, (jobExceptionsResult, jobConfigResult) -> {
return new JobDetails(jobDetailResult, jobExceptionsResult, jobConfigResult);
});
});
}).collect(Collectors.toList());
});
return null;
我的问题是当 JobInfo 为 `new JobInfo(jobsResult, jobDetails) 时如何在这里创建 CompletableFuture?!
正如我所说,我是新手,也许我的方法不好,有更好的解决方案?
任何想法表示赞赏,谢谢
第一个工作版本:
private CompletableFuture<JobInfo> aggregateJobInfo() {
final CompletableFuture<JobsResult> jobsResultFuture = restClient().getJobs();
final CompletableFuture<List<String>> jobIdsFuture = jobsResultFuture.thenApply(JobInfoCollector::extractJobIds);
final CompletableFuture<List<CompletableFuture<JobDetails>>> jobDetailsFutureListFuture =
jobIdsFuture.thenApply(jobIds -> jobIds.stream().map(jobId -> {
final CompletableFuture<JobDetailResult> jobDetailsResultFuture = restClient().getJobDetails(jobId);
final CompletableFuture<JobExceptionsResult> jobExceptionsFuture = restClient().getJobExceptions(jobId);
final CompletableFuture<JobConfigResult> jobConfigFuture = restClient().getJobConfig(jobId);
return jobDetailsResultFuture.thenCompose(jobDetailResult ->
jobExceptionsFuture.thenCombine(jobConfigFuture, (jobExceptionsResult, jobConfigResult) ->
new JobDetails(jobDetailResult, jobExceptionsResult, jobConfigResult)));
}).collect(Collectors.toList()));
return jobDetailsFutureListFuture.thenCompose(jobDetailsFutures ->
CompletableFuture.allOf(jobDetailsFutures.toArray(
new CompletableFuture[jobDetailsFutures.size()])).thenApply(aVoid ->
jobDetailsFutures.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList())))
.thenApply(jobDetails -> jobsResultFuture.thenApply(jobsResult ->
new JobInfo(jobsResult, jobDetails)))
.join();
}
【问题讨论】:
-
您粘贴的大部分代码似乎与问题无关。你能把它简化为你需要的最小例子吗?
-
问题是如何“映射”上面的未来数据,以便返回 CompletableFuture
。