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

les flux parallèles utilisent la valeur par défaut ForkJoinPool.commonPool qui par défaut a un thread en moins que vous avez des processeurs , comme retourné par Runtime.getRuntime().availableProcessors() (cela signifie que les flux parallèles utilisent tous vos processeurs parce qu'ils utilisent aussi 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 pools disponibles processeur.

cela signifie aussi que si vous avez des flux parallèles imbriqués ou plusieurs flux parallèles commencés simultanément, ils vont tous partager 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" attribué à chaque courant parallèle vous lancer (si vous avez plus d'un). (Apparemment vous pouvez utiliser un ManagedBlocker pour contourner cela.)

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

• soumettez l'exécution en flux parallèle à votre propre ForkJoinPool: yourFJP.submit(() -> stream.parallel().forEach(soSomething)).get(); ou
• vous pouvez changer la taille du pool commun en utilisant les propriétés du système: System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20") pour un parallélisme cible de 20 threads.

exemple de la ce dernier sur ma machine qui dispose de 8 processeurs. Si je lance 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 316 316 316 316 316 316 316 415 416 416 416

donc vous pouvez voir que le flux parallèle traite 8 éléments à la fois, i.e. 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

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éclencher le calcul parallèle à l'intérieur de votre propre forkJoinPool, vous pouvez également passer cette piscine à L'avenir completable.méthode 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
);

utiliser un ForkJoinPool et soumettre pour un flux parallèle n'utilise pas de manière fiable tous les threads. Si vous regardez ceci ( le flux parallèle d'un HashSet ne court pas en parallèle ) et ceci ( pourquoi le flux parallèle n'utilise-t-il pas tous les fils du ForkJoinPool? ), vous verrez le raisonnement.

version courte: si ForkJoinPool / submit ne fonctionne pas pour vous, utiliser

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "10");

Jusqu'à présent, j'ai utilisé les solutions décrites dans les réponses de 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 ya des inconvénients en ce qui concerne le mécanisme de division des flux parallèles qui dépend fortement de la taille de la piscine commune. Voir flux parallèle à partir d'un HashSet ne fonctionne pas en parallèle .

j'utilise cette solution seulement pour avoir des piscines séparées pour différents types de travail mais je ne peux pas définir la taille de la piscine commune à 1 même si Je ne l'utilise pas.

pour mesurer le nombre réel de threads utilisés, Vous pouvez cocher 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 CPU 4-core une sortie comme:

5 // common pool
23 // custom pool

sans .parallel() il donne:

3 // common pool
4 // custom pool

Aller pour obtenir de l' AbacusUtil . Le nombre de Thread peut être spécifié pour le flux parallèle. Voici le code échantillon:

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

Divulgation： je suis le développeur de AbacusUtil.

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 poêles sur mesure. 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 voulions poursuivre 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);

[révélation je suis le développeur principal de cyclope-réagir]

j'ai essayé le personnalisé ForkJoinPool 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 disant 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 weirdo , quand j'ai essayé d'atteindre le même résultat en utilisant ThreadPoolExecutor comme 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 démarrera seulement le parallelStream dans un nouveau fil et puis tout le reste est exactement le même, ce qui à nouveau prouve que le parallelStream utilisera le ForkJoinPool pour démarrer ses fils d'enfant.

Note: Il semble y avoir une correction implémentée dans JDK 10 qui assure que le Pool de threads personnalisé utilise le nombre attendu de threads.

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