Pool de threads personnalisé dans le flux parallèle Java 8

Est-il possible de spécifier un pool de threads personnalisé pour le flux parallèle Java 8 ? Je ne peux pas le trouver nulle part.

Imaginez que j'ai une application serveur et que j'aimerais utiliser des flux parallèles. Mais l'application est grande et multithread donc je veux la compartimenter. Je ne veux pas d'une tâche d'exécution lente dans un module des tâches de blocage d'application d'un autre module.

Si je ne peux pas utiliser différents pools de threads pour différents modules, cela signifie que je ne peux pas utiliser en toute sécurité des flux parallèles dans la plupart des situations du monde réel.

Essayez l'exemple suivant. Certaines tâches gourmandes en CPU sont exécutées dans des threads séparés. Les tâches exploitent des flux parallèles. La première tâche est interrompue, chaque étape dure donc 1 seconde (simulée par le sommeil du thread). Le problème est que d'autres threads se bloquent et attendent que la tâche interrompue se termine. C'est un exemple artificiel, mais imaginez une application de servlet et quelqu'un soumettant une tâche de longue durée au pool de jointure de fourches partagé.

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

Il y a en fait une astuce pour exécuter une opération parallèle dans un pool de jointure spécifique. Si vous l'exécutez en tant que tâche dans un pool de jointures de fourches, il y reste et n'utilise pas le plus commun.

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

L'astuce est basée sur ForkJoinTask.fork qui spécifie: "Arrange pour exécuter de manière asynchrone cette tâche dans le pool dans lequel la tâche en cours s'exécute, le cas échéant, ou en utilisant ForkJoinPool.commonPool () sinon dansForkJoinPool ()"

Les flux parallèles utilisent la valeur par défaut ForkJoinPool.commonPoolqui, par défaut, a un thread de moins car vous avez des processeurs , comme retourné par Runtime.getRuntime().availableProcessors()(Cela signifie que les flux parallèles utilisent tous vos processeurs car ils utilisent également le thread principal):

Pour les applications qui nécessitent des pools séparés ou personnalisés, un ForkJoinPool peut être construit avec un niveau de parallélisme cible donné; par défaut, égal au nombre de processeurs disponibles.

Cela signifie également que si vous avez des flux parallèles imbriqués ou plusieurs flux parallèles démarrés simultanément, ils partageront tous le même pool. Avantage: vous n'utiliserez jamais plus que la valeur par défaut (nombre de processeurs disponibles). Inconvénient: vous ne pouvez pas obtenir "tous les processeurs" affectés à chaque flux parallèle que vous lancez (si vous en avez plusieurs). (Apparemment, vous pouvez utiliser un ManagedBlocker pour contourner cela.)

Pour modifier la façon dont les flux parallèles sont exécutés, vous pouvez soit

• soumettre l'exécution du flux parallèle à votre propre ForkJoinPool: yourFJP.submit(() -> stream.parallel().forEach(soSomething)).get();ou
• vous pouvez modifier la taille du pool commun à l'aide des propriétés système: System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20")pour un parallélisme cible de 20 threads. Cependant, cela ne fonctionne plus après le correctif rétroporté https://bugs.openjdk.java.net/browse/JDK-8190974 .

Exemple de ce dernier sur ma machine qui dispose de 8 processeurs. Si j'exécute le programme suivant:

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

La sortie est:

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

Vous pouvez donc voir que le flux parallèle traite 8 éléments à la fois, c'est-à-dire qu'il utilise 8 threads. Cependant, si je décommente la ligne commentée, la sortie est:

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

Cette fois, le flux parallèle a utilisé 20 threads et les 20 éléments du flux ont été traités simultanément.

Alternativement à l'astuce de déclenchement du calcul parallèle dans votre propre forkJoinPool, vous pouvez également passer ce pool à la méthode CompletableFuture.supplyAsync comme dans:

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 solution d'origine (définition de la propriété de parallélisme commun ForkJoinPool) ne fonctionne plus. En regardant les liens dans la réponse d'origine, une mise à jour qui rompt cela a été renvoyée vers Java 8. Comme mentionné dans les threads liés, cette solution n'était pas garantie de fonctionner indéfiniment. Sur cette base, la solution est la solution forkjoinpool.submit with .get discutée dans la réponse acceptée. Je pense que le backport corrige également le manque de fiabilité de cette solution.

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

Nous pouvons changer le parallélisme par défaut en utilisant la propriété suivante:

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

qui peut configurer pour utiliser plus de parallélisme.

Pour mesurer le nombre réel de threads utilisés, vous pouvez vérifier 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();

Cela peut produire sur un processeur à 4 cœurs une sortie comme:

5 // common pool
23 // custom pool

Sans .parallel()cela donne:

3 // common pool
4 // custom pool

Jusqu'à présent, j'ai utilisé les solutions décrites dans les réponses à cette question. Maintenant, j'ai trouvé une petite bibliothèque appelée Parallel Stream Support pour cela:

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

Mais comme @PabloMatiasGomez l'a souligné dans les commentaires, il existe des inconvénients concernant le mécanisme de fractionnement des flux parallèles qui dépend fortement de la taille du pool commun. Voir Le flux parallèle à partir d'un HashSet ne s'exécute pas en parallèle .

J'utilise cette solution uniquement pour avoir des pools distincts pour différents types de travail, mais je ne peux pas définir la taille du pool commun à 1 même si je ne l'utilise pas.

Remarque: Il semble y avoir un correctif implémenté dans JDK 10 qui garantit que le pool de threads personnalisé utilise le nombre attendu de threads.

L'exécution de flux parallèle dans un ForkJoinPool personnalisé doit obéir au parallélisme https://bugs.openjdk.java.net/browse/JDK-8190974

J'ai essayé le ForkJoinPool personnalisé comme suit pour ajuster la taille de la piscine:

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

Voici la sortie indiquant que le pool utilise plus de threads que la valeur par défaut 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]

Mais en fait, il y a un bizarre , quand j'ai essayé d'obtenir le même résultat en utilisant ThreadPoolExecutorcomme suit:

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

mais j'ai échoué.

Il ne démarrera le parallelStream que dans un nouveau thread, puis tout le reste sera le même, ce qui prouve à nouveau que parallelStreamva utiliser ForkJoinPool pour démarrer ses threads enfants.

Allez chercher AbacusUtil . Le numéro de thread peut être spécifié pour le flux parallèle. Voici l exemple de code:

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

Divulgation ： Je suis le développeur d'AbacusUtil.

Si vous ne voulez pas vous fier aux hacks d'implémentation, il existe toujours un moyen d'y parvenir en implémentant des collecteurs personnalisés qui combineront mapet collectsémantiques ... et vous ne serez pas limité à ForkJoinPool:

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

Heureusement, c'est déjà fait ici et disponible sur Maven Central: http://github.com/pivovarit/parallel-collectors

Avertissement: je l'ai écrit et j'en assume la responsabilité.

Si cela ne vous dérange pas d'utiliser une bibliothèque tierce, avec cyclops-react vous pouvez mélanger des flux séquentiels et parallèles dans le même pipeline et fournir des ForkJoinPools personnalisés. Par exemple

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

Ou si nous souhaitions continuer le traitement dans un flux séquentiel

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

[Divulgation Je suis le développeur principal de cyclops-react]

Si vous n'avez pas besoin d'un ThreadPool personnalisé mais que vous souhaitez plutôt limiter le nombre de tâches simultanées, vous pouvez utiliser:

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 question en double demandant cela est verrouillée, alors veuillez me porter ici)

vous pouvez essayer d'implémenter ce ForkJoinWorkerThreadFactory et l'injecter à la classe 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();
    }

vous pouvez utiliser ce constructeur de pool Fork-Join pour ce faire.

notes: - 1. si vous utilisez ceci, tenez compte du fait qu'en fonction de votre implémentation de nouveaux threads, la planification à partir de la JVM sera affectée, ce qui planifie généralement les threads de jointure en fourche sur différents cœurs (traités comme un thread de calcul). 2. La planification des tâches par fork-join aux threads ne sera pas affectée. 3. Je n'ai pas vraiment compris comment le flux parallèle sélectionne les threads à partir de la jointure en fourche (impossible de trouver la documentation appropriée), alors essayez d'utiliser une autre usine threadNaming afin de vous assurer que si les threads du flux parallèle sont sélectionnés de customThreadFactory que vous fournissez. 4. commonThreadPool n'utilisera pas cette customThreadFactory.