我如何在这个 for 循环中使用并行流

Question

jsonarray 对象包含 50000 个 jsonObject。我需要处理这个 jsonObject 来制作一个对象，然后我想把这个对象添加到一个列表中。这需要很多时间，所以我想使用并行流来制作对象并更快地添加到列表中。

这是我想用并行循环替换的代码

        // I want to insert object in this array
        List<Curriculum> curriculamList = new ArrayList<>();

        // This is the org.json array
        JSONArray jsonarray = new JSONArray(content);

        // This loop I want to replace
        for (int i = 0; i < jsonarray.length(); i++) {
            JSONObject jsonobject = jsonarray.getJSONObject(i);

            // Start of processing
            Program program = programDAO.getProgramDetails(jsonobject.getInt("programId"));
            Batch batch = batchDAO.getBatchDetails(jsonobject.getInt("batchId"), program);
            if(batch.getBatchId() == 0)
                continue;
            Major major = majorDAO.getMajorDetails(jsonobject.getInt("majorId"));
            Course course = courseDAO.getCourseDetails(jsonobject.getString("courseCode"));
            if(course == null)
                continue;
            double credits = jsonobject.getDouble("credits");
            CourseType courseType = courseTypeDAO.getCourseTypeDetails(program, jsonobject.optString("type"));
            int semesterCount = jsonobject.getInt("semester");
            String prereqCourseString = jsonobject.getString("prereq");
            Course alternateCourse = courseDAO.getCourseDetails(jsonobject.getString("alternate"));

            List<Course> prereqCourseList = new ArrayList<>();
            if (prereqCourseString.length() != 0) {
                String[] prereqCourseSplit = prereqCourseString.split("AND");
                for (String prereqCourseSplitString : prereqCourseSplit) {
                    prereqCourseList.add(courseDAO.getCourseDetails(prereqCourseSplitString.trim()));
                }
            }
            List<Course> prereqChainCourseList = new ArrayList<>();
            // End of processing

            // This is the object
            Curriculum curriculum = new Curriculum(course, credits, courseType, semesterCount, prereqCourseList, prereqChainCourseList, alternateCourse, batch, major);

            // Pushing object into the list
            curriculamList.add(curriculum);
        }

在这里我尝试了一些代码，但可能是我走错了路。如何在流中使用 list 并获取循环的索引号。请帮助我如何将其转换为流。

        JSONArray jsonarray = new JSONArray(content);
        Stream.of(jsonarray)
                .parallel()
                .forEach(objects -> {
                    // How I can get index number here and push it to the list?
                    JSONObject jsonobject = objects.getJSONObject(i);

                    // Here is the processing code

                    Curriculum curriculum = new Curriculum(course, credits, courseType, semesterCount, prereqCourseList, prereqChainCourseList, alternateCourse, batch, major);

                    // Variable used in lambda expression should be final or effectively final? How to add then?
                    curriculamList.add(curriculum);
                });

我是java新手，如有错误请见谅。

Answer 1

Stream.of(jsonarray) 返回一个带有单个元素的 Stream，即 JSONArray 对象，这不是您想要的。

我认为这就是你想要的：

List<Curriculum> curriculumList = IntStream.range(0, jsonarray.length()).parallel()
            .mapToObj(i -> {
                JSONObject jsonobject = jsonarray.getJSONObject(i);

                // fetch the various parts...

                return new Curriculum(course, credits, courseType, semesterCount, prereqCourseList, prereqChainCourseList, alternateCourse, batch, major);
            }).collect(Collectors.toList());

但是，您的代码似乎进行了多次阻塞网络调用。通常不建议使用并行流来并行执行阻塞网络调用，因为并行流旨在用于 CPU 并行化，因此请使用只有少量线程的单个共享线程池。因此，您可能更喜欢这样的东西：

ExecutorService executor = Executors.newFixedThreadPool(10);

// start many calls running in parallel
List<Future<Curriculum>> futures = IntStream.range(0, jsonarray.length())
        .mapToObj(i -> executor.submit(() -> {
                JSONObject jsonobject = jsonarray.getJSONObject(i);

                // fetch the various parts...

                return new Curriculum(course, credits, courseType, semesterCount, prereqCourseList, prereqChainCourseList, alternateCourse, batch, major);
            })).collect(Collectors.toList());

executor.shutdown();

List<Curriculum> curriculumList = new ArrayList<>();
for (Future<Curriculum> future : futures) {
    curriculumList.add(future.get());
}