java - 将两个 CompletableFutures 链接在一起，每个 CompletableFutures 返回列表

Question

我很难弄清楚这一点，并且可以向那些比我更有经验和知识的人寻求帮助。

基本问题是我需要获取对象列表，然后对于返回的每个对象，获取一些详细信息，并将详细信息缝合到对象中。我希望在这方面保持高效；我需要首先获取数据文件列表，但是一旦有了它，我就可以同时调用来获取它们的所有标签，然后等待所有 getTags 响应返回，然后再处理它们。

public class DataFile {
    // DataFileDao returns all DataFile properties, except Tags
    private List<Tags> tags;
    ...
}

我只是不知道如何使用 CompletableFutures 和流来实现这一功能。不过，这是我正在使用的基本代码，如果有人可以帮助我完成任务，我将不胜感激:

public CompletableFuture<List<DataFile>> getDataFilesWithTags() {

    final CompletableFuture<List<DataFile>> dataFileFutures = this.dataFileDao.getDataFiles()
        .thenApply(HttpResponse::body).thenApply(this::toDataFileList);

    final CompletableFuture<List<List<Tag>>> tagFutures = dataFileFutures
        .thenCompose(dataFiles -> HttpUtils.allAsList(dataFiles.stream()
            .map(file -> this.tagDao.getLatestTagsForDataFile(file.getId())).collect(toList())));

    final CompletableFuture<List<DataFile>> filesWithTags = dataFileFutures.thenCombine(tagFutures,
        (files, tags) -> {
            for (int i = 0; i < files.size(); i++) {
                files.get(i).setTags(tags.get(i));
            }

            return files;
        });

    return fileWithTags;
}

/**
 * Transforms a generic {@link List} of {@link CompletableFuture}s to a {@link CompletableFuture} containing a
 * generic {@link List}.
 *
 * @param futures the {@code List} of {@code CompletableFuture}s to transform
 * @param         <T> the type of {@link CompletableFuture} to be applied to the {@link List}
 * @return a {@code CompletableFuture} containing a {@code List}
 * @throws NullPointerException if {@code futures} is null
 */
public static <T> CompletableFuture<List<T>> allAsList(final List<CompletableFuture<T>> futures) {
    Validate.notNull(futures, "futures cannot be null");
    return CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
            .thenApply(ignored -> futures.stream().map(CompletableFuture::join).collect(Collectors.toList()));
}

必须有一种更干净、更实用的方法来做到这一点，对吗？

我想做的事情的抽象表示:

public class ParentObject {

    // RETURNED BY ParentObjectDao.getAllParentObjects()
    private String id;

    // *NOT* RETURNED BY ParentObjectDao.getAllParentObjects()
    // MUST BE RETRIEVED BY MAKING SUPPLEMENTAL CALL TO ParentObjectDao.getParentsChildren(String parentObjectId)
    private List<ChildObject> details;
}

public class ChildObject {

    private String id;
    private String description;
}

public class ParentObjectDao {

    public CompletableFuture<List<ParentObject>> getAllParentObjects();

    public CompletableFuture<List<ChildObject>> getChildrenForParent(String parentObjectId);
}

public class Something {

    private final ParentObjectDao dao;

    public List<ParentObject> getParentObjectsWithChildren(){

        // PSEUDO-LOGIC
        final List<ParentObject> parentsWithChildren = dao.getAllParentObjects()
            .thenApply(List::stream)
            .thenCompose(parentObject -> dao.getChildrenForParent(parentObject.getId()))
            .thenApply(parentObject::setChildren)
            .collect(toList);

        return parentsWithChildren;
    }
}

Answer 1

您拥有的代码实际上并没有太多并行化。您一次仅处理一个 CompletableFuture，将操作链接到它。因此，如果您有 1000 个数据文件，它们仍然会按顺序处理。

此外，从设计和可读性的角度来看，CompletableFuture 的运行级别太低(您真的需要链接 thenApply(HttpResponse::body). thenApply(this::toDataFileList) 不能正确封装转换并让 CompletableFuture 仅代表一种方法吗？)

使用你的伪代码，像这样的东西怎么样:

CompletableFuture<List<ParentObject>> populateAsync(List<ParentObject> parents) {

  //get the children of each parent in parallel, store the futures in a list
  List<CompletableFuture<ParentObject>> futures = 
    parents.stream() 
           .map(parent ->
                   parentObjectDao.getChildrenForParent(parent.getId())
                                  .thenApply(parent::setChildren))  //assuming setChildren returns the parent object
           .collect(Collectors.toList()); //we need this stream terminal operation to start all futures before we join on the first one

  //wait for all of them to finish and then put the result in a list
  return CompletableFuture.supplyAsync(() -> 
                                     futures.stream()
                                            .map(CompletableFuture::join)
                                            .collect(Collectors.toList());    
}

然后你就可以做这样的事情:

CompletableFuture<List<ParentObject>> getAllParentObjects()
          .thenApply(this::populateAsync)

(我可能有一些语法错误，因为我只是直接写在这里，但你应该明白)。