java - 如何在Java中使用CompletableFuture处理列表列表？

Question

我在Java中有一个List<List<String>>，我想与固定线程池示例3异步处理父列表中的列表。我试图在Java 8中使用CompletableFuture和Stream。到目前为止，我尝试过的PFB代码。在处理器中，我只是打印它，但是我将执行数据库操作。

所以在这里，我尝试流List<List<String>>并根据列表大小创建线程数，但是将流列表作为参数传递给具有CompletableFuture的Processor。

public class CompletableFutureWithList {
    public static void main(String args[]) {
        List<List<String>> aList = new ArrayList<>();
        aList.add(new ArrayList<>(Arrays.asList("xyz", "abc")));
        aList.add(new ArrayList<>(Arrays.asList("qwe", "poi")));
        System.out.println("helo...");
        ExecutorService executor = Executors.newFixedThreadPool(aList.size());
        //aList.stream().flatMap(List::stream).
        Processor aProcessor = new Processor();
        List<String> tempList = new ArrayList<>();
        CompletableFuture aComFuture = supplyAsync(() -> aProcessor.processList(tempList), executor);
        try {
            aComFuture.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}
public class Processor {
    public boolean processList(List<String> tempList) {
        for (String string : tempList) {
            System.out.println("Output: " + string);
        }
        return true;
    }
}  

Answer 1

因此，据我了解，您需要为List<String>中的每个List<List<String>>调用处理器

因此，您可以做的是使用CompletableFuture创建所有新线程，然后等待它们全部完成并对返回值进行任何处理。

所以你可以做的是这样的

List<List<String>> aList = new ArrayList<>();

//Create all CFs
List<CompletableFuture<Boolean>> futureList = aList.stream()
            .map(strings -> CompletableFuture.supplyAsync(() -> processList(strings), executor))
            .collect(toList());

//Wait for them all to complete
CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).join();

//Do processing of the results
Stream<Boolean> booleanStream = futureList.stream()
            .map(CompletableFuture::join);
//Do other stuff you need