通过 Streams 并行执行多个查询

Question

我有以下方法：

public String getResult() {

        List<String> serversList = getServerListFromDB();

        List<String> appList = getAppListFromDB();

        List<String> userList = getUserFromDB();

        return getResult(serversList, appList, userList);
    }

在这里，我依次调用三个方法，依次访问数据库并获取结果，然后我对从数据库访问中获得的结果进行 post 处理。我知道如何通过使用 Threads 同时调用这三种方法。但是我想用 Java 8 Parallel Stream 来实现这个。有人可以指导我如何通过并行流实现相同的目标吗？

编辑我只想通过 Stream 并行调用这些方法。

private void getInformation() {
    method1();
    method2();
    method3();
    method4();
    method5();
}

Answer 1

不是很清楚你的意思，但如果你只是想运行并行处理这些列表，你可以这样做：

    List<String> list1 = Arrays.asList("1", "234", "33");

    List<String> list2 = Arrays.asList("a", "b", "cddd");

    List<String> list3 = Arrays.asList("1331", "22", "33");

    List<List<String>> listOfList = Arrays.asList(list1, list2, list3);

    listOfList.parallelStream().forEach(list -> System.out.println(list.stream().max((o1, o2) -> Integer.compare(o1.length(), o2.length()))));

（它将打印每个列表中最长的元素）。

Answer 2

foreach 用于 side-effects，您可以在 parallel stream 上调用 foreach。例如：

listOfTasks.parallelStream().foreach(list->{
  submitToDb(list);
});

然而，parallelStream 使用常见的 ForkJoinPool，这可以说不适合 IO-bound 任务。

考虑使用 CompletableFuture 并提供适当的 ExecutorService。它提供了更大的灵活性（continuation，配置）。例如：

ExecutorService executorService = Executors.newCachedThreadPool();
List<CompletableFuture> allFutures = new ArrayList<>();
for(Query query:queries){
 CompletableFuture<String> query = CompletableFuture.supplyAsync(() -> {
        // submit query to db
        return result;
  }, executorService);
  allFutures.add(query);
}

 CompletableFuture<Void> all = CompletableFuture.allOf(allFutures.toArray(new CompletableFuture[allFutures.size()]));

Answer 3

如前所述，标准并行流可能不是最适合您的用例。我将使用 ExecutorService 异步完成每个任务，并在调用 getResult 方法时 "join" 它们：

ExecutorService es = Executors.newFixedThreadPool(3);

Future<List<String>> serversList = es.submit(() -> getServerListFromDB());
Future<List<String>> appList = es.submit(() -> getAppListFromDB());
Future<List<String>> userList = es.submit(() -> getUserFromDB());

return getResult(serversList.get(), appList.get(), userList.get());

Answer 4

您可以这样使用 CompletableFuture：

public String getResult() {

    // Create Stream of tasks:
    Stream<Supplier<List<String>>> tasks = Stream.of(
            () -> getServerListFromDB(),
            () -> getAppListFromDB(),
            () -> getUserFromDB());

    List<List<String>> lists = tasks
         // Supply all the tasks for execution and collect CompletableFutures
         .map(CompletableFuture::supplyAsync).collect(Collectors.toList())
         // Join all the CompletableFutures to gather the results
         .stream()
         .map(CompletableFuture::join).collect(Collectors.toList());

    // Use the results. They are guaranteed to be ordered in the same way as the tasks
    return getResult(lists.get(0), lists.get(1), lists.get(2));
}

通过 Streams 并行执行多个查询

Execute multiple queries in parallel via Streams

java

parallel-processing

java-8

java-stream