Comment puis-je utiliser des pools de threads personnalisés avec des flux parallèles Java 8 ?

Pools de threads personnalisables dans Java 8 Flux parallèles

Le contrôle de l'allocation des ressources de thread au sein des flux parallèles améliore les performances et la modularité des applications. Cependant, on ne sait toujours pas comment désigner des pools de threads personnalisés pour des opérations de flux spécifiques dans Java 8.

Cet article explore le problème et propose une solution, surmontant la limitation des pools de threads partagés dans les flux parallèles.

Énoncé du problème

Considérez une application serveur utilisant des flux parallèles. Pour éviter que les tâches lentes au sein d'un module n'aient un impact sur les autres, une compartimentation du pool de threads est souhaitée. Cependant, l'implémentation standard du flux parallèle utilise un pool de threads partagé pour toutes les opérations.

Comme l'illustre l'extrait de code suivant, une tâche compromise retarde indéfiniment les autres threads :

// CPU-intensive tasks in separate threads leveraging parallel streams
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));

Solution

Pour exécuter des opérations parallèles au sein d'un pool de jointure fork désigné, utilisez-le comme tâche au sein du piscine. Cela isole l'opération du pool partagé :

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

Cette technique exploite le comportement de ForkJoinTask.fork() : "Exécutez de manière asynchrone cette tâche dans le pool dans lequel la tâche actuelle s'exécute, le cas échéant, ou en utilisant le ForkJoinPool.commonPool() s'il n'est pas dans ForkJoinPool()."

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!