Как (глобально) заменить общий пул потоков потока параллельных потоков Java?

Я хотел бы глобально заменить общий пул потоков, используемый по умолчанию параллельными потоками Java, то есть, например, для

IntStream.range(0,100).parallel().forEach(i -> {
    doWork();
});

Я знаю, что можно использовать выделенный ForkJoinPool, отправив такую ​​инструкцию в выделенный пул потоков (см. Пользовательский пул потоков в параллельном потоке Java 8). Вопрос здесь

  • Возможно ли заменить обычный ForkJoinPool на некоторую другую реализацию (скажем, Executors.newFixedThreadPool(10)?
  • Можно ли сделать это с помощью некоторых глобальных настроек, например, некоторого свойства JVM?

Примечание. Причина, по которой мне нравится заменять пул F/J, заключается в том, что она, похоже, имеет ошибку, которая делает ее непригодной для вложенных параллельных циклов.

Вложенные параллельные петли имеют низкую производительность и могут приводить к взаимоблокировкам, см. http://christian-fries.de/blog/files/2014-nested-java-8-parallel-foreach.html

Например: следующий код приводит к тупику:

// Outer loop
IntStream.range(0,24).parallel().forEach(i -> {

    // (omitted:) do some heavy work here (consuming majority of time)

    // Need to synchronize for a small "subtask" (e.g. updating a result)
    synchronized(this) {
        // Inner loop (does s.th. completely free of side-effects, i.e. expected to work)
        IntStream.range(0,100).parallel().forEach(j -> {
            // do work here
        });
    }
});

(даже без дополнительного кода в "do work here", учитывая, что parallelism установлен в < 12).

Мой вопрос - как заменить FJP. Если вам нравится обсуждать вложенные параллельные циклы, вы можете проверить Вложенная Java 8 параллельная для каждого цикла, выполняющая бедные. Ожидается ли такое поведение?.

Ответы

Ответ 1

Я думаю, что не так, как предполагается использовать поток API. Кажется, вы используете (mis), используя его для простого выполнения параллельного выполнения задачи (фокусируя внимание на задаче, а не на данных), вместо того, чтобы выполнять параллельную обработку потока (фокусируясь на данных в потоке). Ваш код каким-то образом нарушает некоторые из основных принципов для потоков. (Я пишу "как-то", потому что это действительно не запрещено, но обескуражен): Избегайте состояний и побочных эффектов.

Помимо этого (или, возможно, из-за побочных эффектов), вы используете сильную синхронизацию в своем внешнем цикле, что все остальное, но безвредно!

Хотя в документации не упоминается, параллельные потоки используют общий ForkJoinPool внутри. Независимо от того, является ли это недостатком документации, мы должны просто принять этот факт. JavaDoc ForkJoinTask гласит:

Можно определить и использовать ForkJoinTasks, которые могут блокироваться, но для этого требуются еще три соображения: (1) Завершение нескольких, если какие-либо другие задачи должны зависеть от задачи, которая блокирует внешнюю синхронизацию или ввод-вывод. В эту категорию часто попадают асинхронные задачи типа событий, которые никогда не соединяются (например, эти подклассы CountedCompleter). (2) Чтобы минимизировать воздействие ресурсов, задачи должны быть небольшими; идеально выполняя только (возможно) блокирующее действие. (3) Если API ForkJoinPool.ManagedBlocker не используется или количество возможных заблокированных задач, как известно, меньше, чем уровень пула ForkJoinPool.getParallelism, пул не может гарантировать, что достаточное количество потоков будет доступно для обеспечения прогресса или хорошей производительности.

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

  • Если вы просто хотите выполнять задачи n параллельно, используйте ExecutionService
  • Если у вас более сложный пример, когда задачи создают подзадачи, используйте вместо этого ForkJoinPool (с ForkJoinTasks). (Он обеспечивает постоянное количество потоков без опасности тупика из-за слишком многих задач, ожидающих завершения других, поскольку задачи ожидания не блокируют их исполняемые потоки).
  • Если вы хотите обрабатывать данные (параллельно), подумайте об использовании потока API.
  • Вы не можете установить собственный общий пул. Он создается внутри частного статического кода.
  • Но вы можете влиять на parallelism, поток factory и обработчик исключений из общего пула с использованием определенных свойств системы (см. JavaDoc of ForkJoinPool)

Не смешивайте ExecutionService и ForkJoinPool. Они (обычно) не заменяют друг друга!

Ответ 2

Хотя ваш первоначальный вопрос уже хорошо ответил isnot2bad, для вас может быть важно, чтобы описанная ошибка (причина вашего желания обменяться реализацией FJP) была исправлена ​​с помощью java 1.8.0.40. См. Вложенная Java 8 параллельная для каждого цикла работает плохо. Ожидается ли такое поведение?