Как я могу использовать параллельный поток в этом для цикла

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

Здесь я попробовал какой-то код, но, возможно, я не на том пути. Как я могу использовать список внутри потока и получить порядковый номер цикла. Пожалуйста, помогите, как я могу преобразовать его в поток.

        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());

Однако похоже, что ваш код выполняет несколько блокирующих сетевых вызовов. Часто нежелательно использовать параллельные потоки для параллельного выполнения сетевых вызовов, так как параллельные потоки предназначены для распараллеливания ЦП, поэтому используйте один общий пул потоков, который имеет только небольшое количество потоков. Поэтому вы можете предпочесть что-то вроде этого:

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());

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

executor.shutdown();