Пользовательский пул потоков в параллельном потоке Java 8

Можно ли указать пользовательский пул потоков для параллельного потока Java 8 ? Я не могу найти это нигде.

Представьте, что у меня есть серверное приложение, и я хотел бы использовать параллельные потоки. Но приложение большое и многопоточное, поэтому я хочу разделить его. Я не хочу медленное выполнение задачи в одном модуле задач приложения блока из другого модуля.

Если я не могу использовать разные пулы потоков для разных модулей, это означает, что я не могу безопасно использовать параллельные потоки в большинстве реальных ситуаций.

Попробуйте следующий пример. Есть несколько задач с интенсивным использованием процессора, выполняемых в отдельных потоках. Задачи используют параллельные потоки. Первое задание не выполняется, поэтому каждый шаг занимает 1 секунду (имитируется спящий поток). Проблема в том, что другие потоки застревают и ждут, пока не завершится неработающая задача. Это надуманный пример, но представьте себе приложение сервлета и того, кто отправляет долгосрочную задачу в общий пул соединений ветвлений.

public class ParallelTest {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService es = Executors.newCachedThreadPool();

        es.execute(() -> runTask(1000)); //incorrect task
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));
        es.execute(() -> runTask(0));


        es.shutdown();
        es.awaitTermination(60, TimeUnit.SECONDS);
    }

    private static void runTask(int delay) {
        range(1, 1_000_000).parallel().filter(ParallelTest::isPrime).peek(i -> Utils.sleep(delay)).max()
                .ifPresent(max -> System.out.println(Thread.currentThread() + " " + max));
    }

    public static boolean isPrime(long n) {
        return n > 1 && rangeClosed(2, (long) sqrt(n)).noneMatch(divisor -> n % divisor == 0);
    }
}

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

final int parallelism = 4;
ForkJoinPool forkJoinPool = null;
try {
    forkJoinPool = new ForkJoinPool(parallelism);
    final List<Integer> primes = forkJoinPool.submit(() ->
        // Parallel task here, for example
        IntStream.range(1, 1_000_000).parallel()
                .filter(PrimesPrint::isPrime)
                .boxed().collect(Collectors.toList())
    ).get();
    System.out.println(primes);
} catch (InterruptedException | ExecutionException e) {
    throw new RuntimeException(e);
} finally {
    if (forkJoinPool != null) {
        forkJoinPool.shutdown();
    }
}

Уловка основана на ForkJoinTask.fork, который указывает: «Обеспечивает асинхронное выполнение этой задачи в пуле, в котором выполняется текущая задача, если это применимо, или с использованием ForkJoinPool.commonPool (), если не inForkJoinPool ()»

Параллельные потоки используют значение по умолчанию ForkJoinPool.commonPool которое по умолчанию имеет на один меньше, чем у вас процессоров , как это возвращает Runtime.getRuntime().availableProcessors()(это означает, что параллельные потоки используют все ваши процессоры, потому что они также используют основной поток):

Для приложений, которым требуются отдельные или пользовательские пулы, ForkJoinPool может быть создан с заданным целевым уровнем параллелизма; по умолчанию равно количеству доступных процессоров.

Это также означает, что если у вас есть параллельные потоки или несколько параллельных потоков, запущенных одновременно, все они будут использовать один и тот же пул. Преимущество: вы никогда не будете использовать больше, чем по умолчанию (количество доступных процессоров). Недостаток: вы можете не получить «все процессоры», назначенные каждому параллельному потоку, который вы инициируете (если у вас их больше одного). (Очевидно, вы можете использовать ManagedBlocker, чтобы обойти это.)

Чтобы изменить способ выполнения параллельных потоков, вы можете либо

• отправьте выполнение параллельного потока на свой собственный ForkJoinPool: yourFJP.submit(() -> stream.parallel().forEach(soSomething)).get(); или
• Вы можете изменить размер общего пула, используя системные свойства: System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20")для целевого параллелизма 20 потоков. Тем не менее, это больше не работает после исправленного патча https://bugs.openjdk.java.net/browse/JDK-8190974 .

Пример последнего на моей машине, которая имеет 8 процессоров. Если я запускаю следующую программу:

long start = System.currentTimeMillis();
IntStream s = IntStream.range(0, 20);
//System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20");
s.parallel().forEach(i -> {
    try { Thread.sleep(100); } catch (Exception ignore) {}
    System.out.print((System.currentTimeMillis() - start) + " ");
});

Выход:

215 216 216 216 216 216 216 216 315 316 316 316 316 316 316 316 415 416 416 416

Итак, вы можете видеть, что параллельный поток обрабатывает 8 элементов одновременно, то есть использует 8 потоков. Однако, если я раскомментирую закомментированную строку, результат будет:

215 215 215 215 215 216 216 216 216 216 216 216 216 216 216 216 216 216 216 216

На этот раз параллельный поток использовал 20 потоков, и все 20 элементов в потоке были обработаны одновременно.

В качестве альтернативы хитрости запуска параллельных вычислений внутри вашего собственного forkJoinPool вы также можете передать этот пул в метод CompletableFuture.supplyAsync, как показано в:

ForkJoinPool forkJoinPool = new ForkJoinPool(2);
CompletableFuture<List<Integer>> primes = CompletableFuture.supplyAsync(() ->
    //parallel task here, for example
    range(1, 1_000_000).parallel().filter(PrimesPrint::isPrime).collect(toList()), 
    forkJoinPool
);

Исходное решение (установка свойства общего параллелизма ForkJoinPool) больше не работает. Глядя на ссылки в исходном ответе, обновление, которое ломает это, было обратно перенесено на Java 8. Как упоминалось в связанных потоках, это решение не гарантировало работать вечно. Исходя из этого, решение представляет собой forkjoinpool.submit с решением .get, обсуждаемым в принятом ответе. Я думаю, что backport также исправляет ненадежность этого решения.

ForkJoinPool fjpool = new ForkJoinPool(10);
System.out.println("stream.parallel");
IntStream range = IntStream.range(0, 20);
fjpool.submit(() -> range.parallel()
        .forEach((int theInt) ->
        {
            try { Thread.sleep(100); } catch (Exception ignore) {}
            System.out.println(Thread.currentThread().getName() + " -- " + theInt);
        })).get();
System.out.println("list.parallelStream");
int [] array = IntStream.range(0, 20).toArray();
List<Integer> list = new ArrayList<>();
for (int theInt: array)
{
    list.add(theInt);
}
fjpool.submit(() -> list.parallelStream()
        .forEach((theInt) ->
        {
            try { Thread.sleep(100); } catch (Exception ignore) {}
            System.out.println(Thread.currentThread().getName() + " -- " + theInt);
        })).get();

Мы можем изменить параллелизм по умолчанию, используя следующее свойство:

-Djava.util.concurrent.ForkJoinPool.common.parallelism=16

который может быть настроен на использование большего параллелизма.

Чтобы измерить фактическое количество используемых потоков, вы можете проверить Thread.activeCount():

    Runnable r = () -> IntStream
            .range(-42, +42)
            .parallel()
            .map(i -> Thread.activeCount())
            .max()
            .ifPresent(System.out::println);

    ForkJoinPool.commonPool().submit(r).join();
    new ForkJoinPool(42).submit(r).join();

Это может привести к 4-ядерному процессору, например:

5 // common pool
23 // custom pool

Без .parallel()этого дает:

3 // common pool
4 // custom pool

До сих пор я использовал решения, описанные в ответах на этот вопрос. Для этого я разработал небольшую библиотеку под названием « Поддержка параллельного потока» :

ForkJoinPool pool = new ForkJoinPool(NR_OF_THREADS);
ParallelIntStreamSupport.range(1, 1_000_000, pool)
    .filter(PrimesPrint::isPrime)
    .collect(toList())

Но, как отметил @PabloMatiasGomez в комментариях, существуют недостатки в отношении механизма разделения параллельных потоков, который сильно зависит от размера общего пула. См. Параллельный поток из HashSet не работает параллельно .

Я использую это решение только для того, чтобы иметь отдельные пулы для разных типов работы, но я не могу установить размер общего пула равным 1, даже если я его не использую.

Замечания: Похоже, в JDK 10 реализовано исправление, обеспечивающее использование ожидаемого числа потоков в пуле пользовательских потоков.

Параллельное выполнение потока в пользовательском ForkJoinPool должно подчиняться параллелизму https://bugs.openjdk.java.net/browse/JDK-8190974

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

private static Set<String> ThreadNameSet = new HashSet<>();
private static Callable<Long> getSum() {
    List<Long> aList = LongStream.rangeClosed(0, 10_000_000).boxed().collect(Collectors.toList());
    return () -> aList.parallelStream()
            .peek((i) -> {
                String threadName = Thread.currentThread().getName();
                ThreadNameSet.add(threadName);
            })
            .reduce(0L, Long::sum);
}

private static void testForkJoinPool() {
    final int parallelism = 10;

    ForkJoinPool forkJoinPool = null;
    Long result = 0L;
    try {
        forkJoinPool = new ForkJoinPool(parallelism);
        result = forkJoinPool.submit(getSum()).get(); //this makes it an overall blocking call

    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    } finally {
        if (forkJoinPool != null) {
            forkJoinPool.shutdown(); //always remember to shutdown the pool
        }
    }
    out.println(result);
    out.println(ThreadNameSet);
}

Вот вывод о том, что пул использует больше потоков, чем по умолчанию 4 .

50000005000000
[ForkJoinPool-1-worker-8, ForkJoinPool-1-worker-9, ForkJoinPool-1-worker-6, ForkJoinPool-1-worker-11, ForkJoinPool-1-worker-10, ForkJoinPool-1-worker-1, ForkJoinPool-1-worker-15, ForkJoinPool-1-worker-13, ForkJoinPool-1-worker-4, ForkJoinPool-1-worker-2]

Но на самом деле есть чудак , когда я пытался добиться того же результата, используя ThreadPoolExecutorследующее:

BlockingDeque blockingDeque = new LinkedBlockingDeque(1000);
ThreadPoolExecutor fixedSizePool = new ThreadPoolExecutor(10, 20, 60, TimeUnit.SECONDS, blockingDeque, new MyThreadFactory("my-thread"));

но я потерпел неудачу.

Он только запустит параллельный поток в новом потоке, а затем все остальное будет таким же, что еще раз доказывает, что он parallelStreamбудет использовать ForkJoinPool для запуска своих дочерних потоков.

Иди, чтобы получить AbacusUtil . Номер потока может быть указан для параллельного потока. Вот пример кода:

LongStream.range(4, 1_000_000).parallel(threadNum)...

Раскрытие информации: я разработчик AbacusUtil.

Если вы не хотите полагаться на хаки реализации, всегда есть способ добиться того же самого путем реализации пользовательских сборщиков, которые будут комбинировать mapи collectсемантику ... и вы не будете ограничены ForkJoinPool:

list.stream()
  .collect(parallelToList(i -> fetchFromDb(i), executor))
  .join()

К счастью, это уже сделано и доступно на Maven Central: http://github.com/pivovarit/parallel-collectors

Отказ от ответственности: я написал это и беру на себя ответственность за это.

Если вы не возражаете против использования сторонней библиотеки, с помощью cyclops-реагировать вы можете смешивать последовательные и параллельные потоки в одном конвейере и предоставлять собственные ForkJoinPools. Например

 ReactiveSeq.range(1, 1_000_000)
            .foldParallel(new ForkJoinPool(10),
                          s->s.filter(i->true)
                              .peek(i->System.out.println("Thread " + Thread.currentThread().getId()))
                              .max(Comparator.naturalOrder()));

Или, если мы хотим продолжить обработку в последовательном потоке

 ReactiveSeq.range(1, 1_000_000)
            .parallel(new ForkJoinPool(10),
                      s->s.filter(i->true)
                          .peek(i->System.out.println("Thread " + Thread.currentThread().getId())))
            .map(this::processSequentially)
            .forEach(System.out::println);

[Раскрытие Я ведущий разработчик циклоп-реакции]

Если вам не нужен пользовательский ThreadPool, но вы хотите ограничить количество одновременных задач, вы можете использовать:

List<Path> paths = List.of("/path/file1.csv", "/path/file2.csv", "/path/file3.csv").stream().map(e -> Paths.get(e)).collect(toList());
List<List<Path>> partitions = Lists.partition(paths, 4); // Guava method

partitions.forEach(group -> group.parallelStream().forEach(csvFilePath -> {
       // do your processing   
}));

(Дубликат вопроса об этом заблокирован, поэтому, пожалуйста, несите меня сюда)

Вы можете попробовать реализовать этот ForkJoinWorkerThreadFactory и внедрить его в класс Fork-Join.

public ForkJoinPool(int parallelism,
                        ForkJoinWorkerThreadFactory factory,
                        UncaughtExceptionHandler handler,
                        boolean asyncMode) {
        this(checkParallelism(parallelism),
             checkFactory(factory),
             handler,
             asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
             "ForkJoinPool-" + nextPoolId() + "-worker-");
        checkPermission();
    }

Вы можете использовать этот конструктор пула Fork-Join для этого.

примечания: - 1. если вы используете это, примите во внимание, что на основе вашей реализации новых потоков это повлияет на планирование из JVM, которое обычно планирует потоки fork-join к различным ядрам (рассматриваются как вычислительный поток). 2. Планирование задач с помощью fork-join к потокам не пострадает. 3. На самом деле не понял, как параллельный поток выбирает потоки из fork-join (не смог найти соответствующую документацию по нему), поэтому попробуйте использовать другую фабрику threadNaming, чтобы убедиться, что потоки в параллельном потоке выбираются от customThreadFactory, которую вы предоставляете. 4. commonThreadPool не будет использовать этот customThreadFactory.