如何在 Java 中使用 CompletableFuture 处理 List 列表？答案

【问题标题】：How to process List of List with CompletableFuture in java?如何在 Java 中使用 CompletableFuture 处理 List 列表？
【发布时间】：2017-07-22 06:17:00
【问题描述】：

我在 java 中有一个 List<List<String>>，我想使用固定线程池示例 3 异步处理父列表中的列表。我正在尝试在 java 8 中使用 CompletableFuture 和 Stream。我不明白如何合并这两个和如何进行。到目前为止我尝试过的 PFB 代码。在处理器中我只是打印它，但我会做数据库操作。

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

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;
    }
}

【问题讨论】：

为什么要使用 CompletableFuture 而不是简单地为每个列表调用 Future<Boolean> f = excecutor.submit(() -> processList(list))？
您可以编写一个简单的程序，它有一个线程（易于调试）并使用执行外部进程 Apache Commons Exec 来运行该程序。

标签： java multithreading java-8 java-stream completable-future

【解决方案1】：

因此，据我了解，您需要在 List<List<String>> 中为每个 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

【讨论】：

【解决方案2】：

这就是合并 list 和 completablefuture 的方法。

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("hello...");

    Processor aProcessor = new Processor();
    List<String> tempList = new ArrayList<>();
    CompletableFuture aComFuture = CompletableFuture.supplyAsync(() -> "");

    aList.stream()
            .forEach(list -> aComFuture.thenApply(fn -> aProcessor.processList(list)));

    aComFuture.join();
}

static class Processor {
    public boolean processList(List<String> tempList) {
        for (String string : tempList) {
            System.out.println("Output: " + string);
        }
        return true;
    }
}

【讨论】：