Flux parallèles : syntaxe et utilisation

1. Rappel de l’API Stream

Vous connaissez déjà la Stream API — c’est un moyen pratique de travailler avec les collections, qui permet d’écrire un code compact et lisible pour le traitement des données : filtrage, tri, comptage, etc.

Voici un exemple classique :

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);

int sum = numbers.stream()
    .filter(n -> n % 2 == 0)
    .mapToInt(n -> n)
    .sum();

System.out.println(sum); // 6 (2 + 4)

Dans cet exemple, la collection est transformée en flux (stream()), seuls les nombres pairs y sont retenus, ils sont ensuite convertis en int, et le résultat est sommé via l’appel à sum().

La Stream API rend le code plus court et plus expressif : au lieu de décrire étape par étape comment tout se passe, vous indiquez simplement ce que vous voulez obtenir. De plus, si nécessaire, on peut facilement basculer vers un traitement parallèle — en une seule ligne.

2. Flux parallèles : syntaxe et principe de fonctionnement

Comment rendre un flux parallèle ?

C’est simple : au lieu de stream(), utilisez parallelStream(). Ou appelez la méthode .parallel() sur un flux existant.

List<Integer> numbers = ...;

int sum = numbers.parallelStream()
    .filter(n -> n % 2 == 0)
    .mapToInt(n -> n)
    .sum();

Ou ainsi :

numbers.stream()
    .parallel() // on le transforme en flux parallèle
    .filter(...)
    .map(...)
    .sum();

Que se passe‑t‑il sous le capot ?

• La collection est automatiquement découpée en parties.
• Chaque partie est traitée dans un thread distinct (via ForkJoinPool — un pool de threads spécialisé).
• Les résultats sont fusionnés en une valeur finale.

Autrement dit, si vous avez un processeur multicœur, le traitement se fait réellement en parallèle — par exemple, le filtrage et le calcul de la somme peuvent s’exécuter simultanément sur plusieurs cœurs.

Dans quels cas est-ce particulièrement utile ?

• Traitement de grandes collections (des dizaines de milliers d’éléments et plus).
• Calculs complexes pour chaque élément.
• Absence de nécessité de conserver un ordre strict de traitement.

Exemple : comparaison entre flux séquentiel et flux parallèle

Regardons un exemple simple avec le traitement d’un grand tableau.

import java.util.*;
import java.util.stream.*;

public class ParallelStreamDemo {
    public static void main(String[] args) {
        List<Integer> numbers = IntStream.rangeClosed(1, 10_000_000)
                                         .boxed()
                                         .collect(Collectors.toList());

        // Séquentiel
        long time1 = System.currentTimeMillis();
        long count1 = numbers.stream()
            .filter(n -> n % 2 == 0)
            .count();
        long time2 = System.currentTimeMillis();
        System.out.println("Séquentiel : " + (time2 - time1) + " ms, nombres pairs : " + count1);

        // Parallèle
        long time3 = System.currentTimeMillis();
        long count2 = numbers.parallelStream()
            .filter(n -> n % 2 == 0)
            .count();
        long time4 = System.currentTimeMillis();
        System.out.println("Parallèle : " + (time4 - time3) + " ms, nombres pairs : " + count2);
    }
}

Essayez ce code sur votre ordinateur — il est fort probable que le flux parallèle traite la collection plus vite (surtout si vous avez un processeur multicœur). Mais pas toujours ! Les nuances — ci‑dessous.

3. Comment cela fonctionne : ForkJoinPool et découpage automatique

Les flux parallèles utilisent sous le capot ForkJoinPool.commonPool(), qui gère automatiquement le nombre de threads (généralement — selon le nombre de cœurs processeur disponibles).

Schématiquement :

+-----------------------------+
|        Votre collection     |
+-----------------------------+
| 1  | 2  | 3  | ... | 10 M   |
+----+----+----+-----+--------+
   |    |    |           |
   v    v    v           v
[Thread1][Thread2]...[ThreadN]
   |    |    |           |
   +----+----+-----------+
        |
   [Fusion des résultats]

Chaque thread traite sa partie, puis les résultats sont agrégés.

4. Limites et écueils

Les flux parallèles ne sont pas un bouton magique « tout accélérer ». Parfois, ils ralentissent même l’exécution !

Quand le parallélisme n’est pas avantageux :

• La collection est petite (jusqu’à ~1000 éléments).
• L’opération sur chaque élément est très rapide (par exemple simplement n * 2).
• Vous avez besoin d’un ordre strict de traitement (par exemple pour une écriture séquentielle dans un fichier).

Pourquoi ? La création et la synchronisation des threads prennent aussi du temps. Si la tâche elle‑même est « petite », les surcoûts peuvent dépasser le gain du parallélisme.

Les effets de bord sont l’ennemi du parallélisme

Si vos opérations dans le flux modifient des variables externes, soyez prudents !

Mauvais exemple :

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
int[] sum = {0};

numbers.parallelStream().forEach(n -> sum[0] += n);

System.out.println(sum[0]); // ??? (vous attendez 15, mais vous pouvez obtenir n’importe quoi)

Pourquoi ? Parce que plusieurs threads modifient simultanément la même variable — il se produit une race condition (état de concurrence). La valeur finale peut être incorrecte.

La bonne approche consiste à utiliser les méthodes du flux qui renvoient un résultat :

int sum = numbers.parallelStream().mapToInt(n -> n).sum();

Toutes les collections ne se parallélisent pas aussi bien

Certaines collections (par exemple un ArrayList classique) se découpent bien. En revanche, LinkedList ou un flux infini (par exemple Stream.generate(...)) — beaucoup moins.

5. Pratique : comparaison des performances

Exemple : recherche du nombre maximal

import java.util.*;
import java.util.stream.*;

public class ParallelMaxDemo {
    public static void main(String[] args) {
        List<Integer> numbers = IntStream.rangeClosed(1, 30_000_000)
                                         .boxed()
                                         .collect(Collectors.toList());

        // Séquentiel
        long t1 = System.currentTimeMillis();
        int max1 = numbers.stream().max(Integer::compareTo).get();
        long t2 = System.currentTimeMillis();
        System.out.println("Séquentiel : " + (t2 - t1) + " ms, max = " + max1);

        // Parallèle
        long t3 = System.currentTimeMillis();
        int max2 = numbers.parallelStream().max(Integer::compareTo).get();
        long t4 = System.currentTimeMillis();
        System.out.println("Parallèle : " + (t4 - t3) + " ms, max = " + max2);
    }
}

Que va‑t‑on observer ? Sur des processeurs multicœurs modernes, le flux parallèle est généralement plus rapide. Mais si vous remplacez 30_000_000 par 1000, la différence disparaît — et parfois le parallèle est même plus lent !

6. Exemples d’utilisation : filtrage, agrégation, tri

Filtrage et comptage

List<String> names = Arrays.asList("Anna", "Boris", "Basil", "Gregory", "Dіana", "George", "Eugene");

long count = names.parallelStream()
    .filter(name -> name.length() == 4)
    .count();

System.out.println("Noms de longueur 4 : " + count);

Regroupement

List<String> words = Arrays.asList("cat", "whale", "cat", "dog", "whale", "cat");

Map<String, Long> freq = words.parallelStream()
    .collect(Collectors.groupingBy(
        w -> w,
        Collectors.counting()
    ));

System.out.println(freq); // {dog=1, whale=2, cat=3}

Tri (mais ici le parallélisme n’apporte pas toujours un gain !)

List<Integer> bigList = IntStream.rangeClosed(1, 5_000_000)
                                 .boxed()
                                 .collect(Collectors.toList());

long t1 = System.currentTimeMillis();
List<Integer> sorted = bigList.parallelStream()
    .sorted()
    .collect(Collectors.toList());
long t2 = System.currentTimeMillis();

System.out.println("Tri parallèle : " + (t2 - t1) + " ms");

7. Points importants et recommandations

Quand utiliser parallelStream()

• La collection est grande (des dizaines de milliers d’éléments et plus).
• L’opération sur l’élément est « lourde » (calculs complexes, travail avec fichiers/réseau).
• Pas de dépendance à l’ordre des éléments.
• Pas d’effets de bord (les variables externes ne sont pas modifiées).

Quand il NE faut PAS utiliser parallelStream()

• La collection est petite.
• L’opération est rapide.
• Il faut conserver strictement l’ordre.
• Il y a un accès à des variables partagées (envisagez des collections thread-safe ou d’autres approches).

Comment savoir combien de threads sont utilisés ?

Par défaut — selon le nombre de cœurs du processeur : Runtime.getRuntime().availableProcessors(). Vous pouvez modifier ce comportement via une propriété système :

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

Ne le faites que si vous en comprenez les conséquences — sinon vous risquez de saturer le processeur et d’obtenir des ralentissements.

8. Erreurs typiques avec les flux parallèles

Erreur n° 1 : Effets de bord à l’intérieur de forEach
Beaucoup pensent : « Maintenant je vais remplir une liste en parallèle ! »

List<Integer> result = new ArrayList<>();
IntStream.range(0, 1_000)
    .parallel()
    .forEach(result::add); // DANGEREUX !
System.out.println(result.size()); // Le résultat est aléatoire !

Pourquoi est‑ce mauvais ? ArrayList n’est pas thread-safe, et lors d’ajouts simultanés depuis plusieurs threads, le résultat est imprévisible : oublis, doublons, exceptions.

Solution : Utilisez les méthodes de collecte du flux (collect), qui garantissent elles‑mêmes la sécurité, ou des collections spécialisées.

List<Integer> result = IntStream.range(0, 1_000)
    .parallel()
    .boxed()
    .collect(Collectors.toList());

Erreur n° 2 : Attendre une accélération sur de petites tâches
Le parallélisme n’est pas gratuit ! Si la collection est petite, le flux parallèle peut fonctionner plus lentement à cause des surcoûts de planification et de synchronisation.

Erreur n° 3 : Violation de l’ordre
Si l’ordre des éléments est important (par exemple lors de l’écriture dans un fichier), n’utilisez pas les flux parallèles — l’ordre n’est pas garanti (ou ce sera plus lent).

Erreur n° 4 : Utiliser des collections « mal adaptées »
Certaines collections (par exemple LinkedList, structures non standard) se découpent mal — l’efficacité du parallélisme diminue.

Erreur n° 5 : Ignorer la sûreté vis‑à‑vis des threads lors de la collecte des résultats
Si vous collectez les résultats manuellement (par exemple en ajoutant dans une liste), utilisez des collections thread-safe (CopyOnWriteArrayList, ConcurrentLinkedQueue) ou les méthodes de collecte du flux.