java - 如何在 Java 8 中从 CompletableFuture> 获取结果

Question

Java 8 环境。

同时使用 CompletableFuture.allOf() 运行任务然后从每个线程获取每个结果，然后将所有结果组合成一个组合结果并返回。

在下面的代码中，要获得结果 ( = List<Student> )，它不必是 I. 和 II 之间的代码。他们说我需要使用 join() 但不起作用

我还从

得到了一些 allOf()

Java 8 CompletableFuture.allOf(...) with Collection or List
和其他链接，但对我来说没有任何作用。我想我错过了一些非常简单的部分。有人知道如何让它发挥作用吗？

public class Test1 {
    public static void main(String[] args) {
        Test1 t = new Test1();
        Map<Major, List<Student>> allMajorStudentListMap = new HashMap<>();
        // fill out some data toallMajorStudentListMap
        t.getData(allMajorStudentListMap);
    }

    List<Student> getData(Map<Major, List<Student>> allMajorStudentListMap) {
        List<CompletableFuture<List<Student>>> completableFutures = new ArrayList<>();

        // suppose the size of completableFutures is 10
        for(Map.Entry<Major, List<Student>> entry: allMajorStudentListMap.entrySet()) {
            CompletableFuture<List<Student>> future = CompletableFuture.supplyAsync(() -> getDetailedStudents(entry));
            completableFutures.add(future);
        }

        // want to run 10 jobs concurrently --> get the 10 result and then combine these 10 results into one
        // finally want to sent the combined 10 results at one in this method

        // I. ======================= I got this code from somewhere     ==========================

        CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0]))
                .exceptionally(ex -> null)
                .join();

        Map<Boolean, List<CompletableFuture<List<Student>>>> result =
                completableFutures.stream()
                        .collect(Collectors.partitioningBy(CompletableFuture::isCompletedExceptionally));

        result.forEach((k, clist) -> {
            System.out.printf("k = " + k);

            for(CompletableFuture<List<Student>> student: clist) {

            // 3) don't know how to get only List<Student> and then print here
            // tried this and that but didn't work
            // student.get() has compile error

            }

        });

        // II. =============================================================================================


        // want to return combined List<Student>
        return ???;
    }

    List<Student> getDetailedStudents(Map.Entry<Major, List<Student>> entry) 
    {
        List<Student> studentList = new ArrayList<>();

        Major major = entry.getKey();
        String majorCode = major.getMajorCode();
        String majorName = major.getMajorName();
        List<Student> studentListList = entry.getValue();           

        studentList.addAll(getDataFromRemote(majorCode, majorName, studentList)));
        return studentList;
    }

    List<Student> getDataFromRemote(String majorCode, String majorName, List<studentList> studentList) {
        // do something and then return list of Student

        return detailedStudentList;
    }
}

Answer 1

在这里，我创建了一个稍作修改的(直接使用 List<Student> 而不是 Map<K,V> ) 版本的工作示例。您可以将您的解决方案与此解决方案进行比较。

总共查询了学生列表五次，每次都同时执行，在人为延迟 3 秒后返回一个完整的学生列表，其中包含一个学生对象。因此，理论上，如果每个对象同时运行，则延迟 3 秒后，所有 5 个学生对象都应该显示出来。

如果您注意到 main 方法的开始和结束之间的时间间隔，大约为 3 秒。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.*;
import java.util.stream.Collectors;

public class CompletableFutureTest {
    private static int counter = 0;
    public static void main(String[] args) {
        System.out.println("Program started at " + new Date());
       List<Student> allStudents = new ArrayList<>();
        new CompletableFutureTest().getData(allStudents);
        for(Student st : allStudents){
            System.out.println(st.getName());
        }
        System.out.println("Program ended at " + new Date());
    }

    private void getData(List<Student> resultToFillIn){
        List<CompletableFuture<List<Student>>> completableFutures = new ArrayList<>();
        //for simulation purpose just running regular for loop 5 times instead of yours          
        final Integer integer = new Integer(0);
        for(int i=0; i < 5; i++){
            completableFutures.add(CompletableFuture.supplyAsync(() -> getStudentsBatch()));
        }
        CompletableFuture<List<Student>>[] cfArray = new CompletableFuture[completableFutures.size()];
        cfArray = completableFutures.toArray(cfArray);
        CompletableFuture.allOf(cfArray)
                .exceptionally(ex ->
                {
                    ex.printStackTrace();
                    return null;
                }).join();
       List<CompletableFuture<List<Student>>> completedFutures = completableFutures.stream().filter(cf -> !cf.isCompletedExceptionally()).collect(Collectors.toList());
       for(CompletableFuture<List<Student>> cf : completedFutures){
           resultToFillIn.addAll(cf.join());
       }
    }

    private List<Student> getStudentsBatch() {
        //adding some delay
        try {
            Thread.sleep( 3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        List<Student> students = new ArrayList<>();
        Student student = new Student();
        student.setName("Student " + ++counter);
        students.add(student);
        return students;
    }

    public static class Student{
        private String name;

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }
    }

    public static class Major{
        private String name;

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }
    }
}

由于其工作原理如上所述，CompletableFuture.allOf(...)正在工作。

但是，尽量避免使用 j oin()只要有可能，因为它会停止当前正在执行的线程。如果您选择真正的异步编程，那么不要使用 join()您可以使用thenAccept(x -> {}) , thenApply(x -> {})回调方法。

希望这对您有帮助。