Grupo de subprocesos personalizado en flujo paralelo de Java 8

¿Es posible especificar un grupo de subprocesos personalizado para la secuencia paralela de Java 8 ? No puedo encontrarlo en ningún lado.

Imagine que tengo una aplicación de servidor y me gustaría usar flujos paralelos. Pero la aplicación es grande y multiproceso, por lo que quiero compartimentarla. No quiero una tarea de ejecución lenta en un módulo de las tareas de bloqueo de aplicaciones de otro módulo.

Si no puedo usar diferentes grupos de subprocesos para diferentes módulos, significa que no puedo usar flujos paralelos de forma segura en la mayoría de las situaciones del mundo real.

Prueba el siguiente ejemplo. Hay algunas tareas intensivas de CPU ejecutadas en subprocesos separados. Las tareas aprovechan flujos paralelos. La primera tarea se interrumpe, por lo que cada paso dura 1 segundo (simulado por el hilo de suspensión). El problema es que otros hilos se atascan y esperan a que termine la tarea rota. Este es un ejemplo artificial, pero imagine una aplicación de servlet y alguien que envíe una tarea de larga ejecución al grupo de unión de fork compartida.

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

En realidad, hay un truco sobre cómo ejecutar una operación paralela en un grupo de bifurcación específico. Si lo ejecuta como una tarea en un grupo fork-join, permanece allí y no usa el común.

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

El truco se basa en ForkJoinTask.fork que especifica: "Organiza la ejecución asincrónica de esta tarea en el grupo en el que se está ejecutando la tarea actual, si corresponde, o utiliza el ForkJoinPool.commonPool () si no está enForkJoinPool ()"

Las secuencias paralelas usan el valor predeterminado, ForkJoinPool.commonPoolque por defecto tiene un subproceso menos, ya que tiene procesadores , como se muestra en Runtime.getRuntime().availableProcessors()(Esto significa que las secuencias paralelas usan todos sus procesadores porque también usan el hilo principal):

Para aplicaciones que requieren agrupaciones separadas o personalizadas, un ForkJoinPool puede construirse con un nivel de paralelismo objetivo dado; por defecto, igual al número de procesadores disponibles.

Esto también significa que si ha anidado flujos paralelos o múltiples flujos paralelos iniciados simultáneamente, todos compartirán el mismo grupo. Ventaja: nunca usará más que el predeterminado (número de procesadores disponibles). Desventaja: es posible que no obtenga "todos los procesadores" asignados a cada flujo paralelo que inicie (si tiene más de uno). (Aparentemente, puedes usar un ManagedBlocker para evitar eso).

Para cambiar la forma en que se ejecutan las transmisiones paralelas, puede

• envíe la ejecución de flujo paralelo a su propio ForkJoinPool: yourFJP.submit(() -> stream.parallel().forEach(soSomething)).get();o
• puede cambiar el tamaño del grupo común utilizando las propiedades del sistema: System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20")para un paralelismo objetivo de 20 hilos. Sin embargo, esto ya no funciona después del parche con respaldo https://bugs.openjdk.java.net/browse/JDK-8190974 .

Ejemplo de esto último en mi máquina que tiene 8 procesadores. Si ejecuto el siguiente programa:

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

El resultado es:

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

Entonces puede ver que la secuencia paralela procesa 8 elementos a la vez, es decir, utiliza 8 subprocesos. Sin embargo, si descomento la línea comentada, el resultado es:

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

Esta vez, la secuencia paralela ha utilizado 20 subprocesos y los 20 elementos de la secuencia se han procesado simultáneamente.

Alternativamente al truco de activar el cálculo paralelo dentro de su propio forkJoinPool, también puede pasar ese grupo al método CompletableFuture.supplyAsync como en:

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

La solución original (establecer la propiedad de paralelismo común de ForkJoinPool) ya no funciona. Mirando los enlaces en la respuesta original, una actualización que rompe esto se ha vuelto a portar a Java 8. Como se mencionó en los hilos enlazados, no se garantiza que esta solución funcione para siempre. Basado en eso, la solución es forkjoinpool.submit con la solución .get discutida en la respuesta aceptada. Creo que el backport también corrige la falta de fiabilidad de esta solución.

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

Podemos cambiar el paralelismo predeterminado usando la siguiente propiedad:

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

que se puede configurar para usar más paralelismo.

Para medir la cantidad real de hilos usados, puede verificar 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();

Esto puede producir en una CPU de 4 núcleos una salida como:

5 // common pool
23 // custom pool

Sin .parallel()eso da:

3 // common pool
4 // custom pool

Hasta ahora, utilicé las soluciones descritas en las respuestas de esta pregunta. Ahora, se me ocurrió una pequeña biblioteca llamada Parallel Stream Support para eso:

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

Pero como @PabloMatiasGomez señaló en los comentarios, hay inconvenientes con respecto al mecanismo de división de las corrientes paralelas que depende en gran medida del tamaño del grupo común. Ver flujo paralelo desde un HashSet no se ejecuta en paralelo .

Estoy usando esta solución solo para tener grupos separados para diferentes tipos de trabajo, pero no puedo establecer el tamaño del grupo común en 1, incluso si no lo uso.

Nota: Parece que hay una solución implementada en JDK 10 que garantiza que el Grupo de subprocesos personalizado use el número esperado de subprocesos.

La ejecución de flujo paralelo dentro de un ForkJoinPool personalizado debe obedecer el paralelismo https://bugs.openjdk.java.net/browse/JDK-8190974

Probé el ForkJoinPool personalizado de la siguiente manera para ajustar el tamaño de la piscina:

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

Aquí está el resultado que dice que el grupo está usando más subprocesos que el predeterminado 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]

Pero en realidad hay un bicho raro , cuando traté de lograr el mismo resultado usando ThreadPoolExecutorlo siguiente:

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

Pero fallé.

Solo iniciará el paralelismo en un nuevo subproceso y luego todo lo demás será igual, lo que nuevamente prueba que parallelStreamutilizará el ForkJoinPool para iniciar sus subprocesos secundarios.

Ve a buscar AbacusUtil . El número de subproceso puede especificarse para flujo paralelo. Aquí está el código de ejemplo:

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

Divulgación: Soy el desarrollador de AbacusUtil.

Si no desea confiar en los hacks de implementación, siempre hay una manera de lograr lo mismo mediante la implementación de recopiladores personalizados que se combinarán mapy la collectsemántica ... y no estaría limitado a ForkJoinPool:

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

Afortunadamente, ya está hecho aquí y disponible en Maven Central: http://github.com/pivovarit/parallel-collectors

Descargo de responsabilidad: lo escribí y asumo la responsabilidad.

Si no le importa usar una biblioteca de terceros, con cyclops-react puede mezclar Streams secuenciales y paralelos dentro de la misma tubería y proporcionar ForkJoinPools personalizados. Por ejemplo

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

O si deseamos continuar procesando dentro de una secuencia secuencial

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

[Divulgación Soy el desarrollador principal de cyclops-react]

Si no necesita un ThreadPool personalizado pero prefiere limitar el número de tareas simultáneas, puede usar:

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

(La pregunta duplicada que pide esto está bloqueada, así que por favor llévame aquí)

puede intentar implementar este ForkJoinWorkerThreadFactory e inyectarlo en la clase 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();
    }

puede usar este constructor del grupo Fork-Join para hacer esto.

notas: 1. si usa esto, tenga en cuenta que, en función de su implementación de nuevos subprocesos, la programación de JVM se verá afectada, que generalmente programa subprocesos de unión de horquilla en diferentes núcleos (tratados como un subproceso computacional). 2. la programación de tareas por fork-join a hilos no se verá afectada. 3. Realmente no he descubierto cómo la secuencia paralela está eligiendo subprocesos de fork-join (no se pudo encontrar la documentación adecuada), así que intente usar una fábrica de nombres de subprocesos diferente para asegurarse de que si se seleccionan subprocesos en secuencia paralela de customThreadFactory que proporciona. 4. commonThreadPool no utilizará este customThreadFactory.