Java 24 – Un aperçu de l’API Gatherer

Depuis son introduction dans JDK 8, l’API Stream a changé profondément la manière dont nous traitons les collections et les séquences de données en Java. Grâce à son approche fonctionnelle et à son support des opérations en pipeline, elle permet d’écrire un code plus expressif et souvent plus efficace.

L’API Stream est très riche et offre de nombreuses possibilités de traitement des données en mémoire, selon le modèle map-filter-reduce. Elle s’appuie sur l’API Spliterator et la polyvalence de ces deux modèles offre de nombreuses possibilités. L’inconvénient est que l’API Spliterator n’est pas facile à utiliser et ne produit pas de code simple et lisible. De plus, si vous devez exploiter des flux parallèles, son utilisation peut s’avérer très complexe.

Avec l’API Gatherer, introduite officiellement dans JDK 24 après plusieurs phases de preview, Java propose un mécanisme pour définir des opérations intermédiaires personnalisées. Cette nouveauté ouvre la porte à une flexibilité et une expressivité inédites dans les pipelines de flux. Sa conception est similaire à celle de l’API Collector pour les opérations Stream terminales.

Comme décrit dans JEP 485, cette API permet d’enrichir les pipelines de flux en introduisant des opérations intermédiaires avancées auparavant non supportées.

Table of Contents

Une solution aux limites des transformations de flux

L’API Stream propose des opérations essentielles telles que map, filter, reduce et sort. Toutefois, certaines transformations avancées restent difficiles à réaliser avec ces outils standards, par exemple :

Appliquer une opération distinct selon un attribut d’un objet (ex. : filtrer des chaînes de caractères par leur longueur unique).
Regrouper des éléments en fenêtres de taille fixe au fil du flux.
Détecter des motifs ou séquences dans un flux d’événements, nécessitant un état persistant.

Ces opérations ne peuvent pas être définies avec l’API Stream et nécessitent de recourir aux Spliterators. L’API Gatherer offre une solution à ces problèmes en permettant de modéliser des opérations intermédiaires sur mesure, de la même manière que l’API Collector permet de définir des opérations terminales personnalisées.

Elle propose un modèle plus simple que l’API Spliterator, avec une excellente prise en charge du parallélisme : même avec une implémentation séquentielle de votre Gatherer, vous pouvez bénéficier des performances offertes par les flux parallèles sur l’amont et l’aval. L’API Spliterator ne vous offre pas cet avantage.

Qu’est-ce qu’un Gatherer ?

Un Gatherer est un objet définissant une opération intermédiaire personnalisée sur un flux. Il reçoit des éléments d’un flux en amont (upstream), effectue une transformation, et transmet (ou non) les éléments à un flux en aval (downstream).

Un Gatherer peut prendre plusieurs formes :

1-to-1 : chaque élément en entrée produit un élément en sortie (comme map).
1-to-N : chaque élément produit zéro ou plusieurs éléments en sortie (comme flatMap).
N-to-1 : plusieurs éléments en entrée sont combinés en un seul (comme fold).
N-to-M : transformation complexe impliquant plusieurs entrées et sorties (ex. : opérations de fenêtrage).

Les Gatherers offrent plusieurs fonctionnalités avancées :

Le maintien d’un état mutable pour conserver une mémoire des opérations précédentes.
L’interruption du traitement pour gérer les flux infinis ou optimiser les traitements (short-circuit).
Le support de l’exécution parallèle via une fonction de combinaison d’état (combiner).

Comprendre l’interface Gatherer

Un Gatherer est défini via l’interface java.util.stream.Gatherer, qui repose sur quatre méthodes clés :

initializer() (optionnel) : fournit un état initial mutable, utilisé pour maintenir des informations à travers les éléments.
integrator() (obligatoire) : définit la transformation à appliquer à chaque élément sous la forme d’un Integrator. Il reçoit l’état courant, l’élément d’entrée et un Downstream pour pousser les éléments transformés.
combiner() (optionnel) : combine les états en cas de traitement parallèle.
finisher() (optionnel) : exécute une opération finale après traitement de tous les éléments.

Un Gatherer peut être instancié en implémentant directement cette interface ou via des méthodes de fabrique comme Gatherer.of() et Gatherer.ofSequential().

Exemple d’utilisation de Gatherer

L’API Gatherer existe pour simplifier des opérations complexes. Il est donc pratiquement impossible d’écrire des exemples simples qui rendent compte de l’intérêt de l’API. Voici toutefois deux exemples triviaux.

map

Cette opération effectue le même travail qu’une opération map et transforme toutes les chaînes de caractères d’un flux en majuscules.

Java

var gatherer = Gatherer.<String, String>of(
  (_, element, downstream) -> downstream.push( element.toUpperCase() )
);

Stream.of( "a", "b", "c" )
  .gather( gatherer )
  .forEach( System.out::println ); // Sortie : A, B, C

Ici, Gatherer.of() est utilisé avec une lambda pour transformer chaque élément en majuscule avant de le transmettre à l’étape suivante. Cette version est la plus simple disponible : elle ne nécessite pas de maintenir un état.

💡Il est important de noter que l’intégrateur doit retourner un boolean qui indique si le flux accepte d’avantage de donnée ou non. Dans cet exemple, nous nous contentons de reporter l’information donnée par l’opération push sur le flux aval, faisant ainsi remonter une éventuelle fin de traitement vers l’amont (short-circuit).

join

Le Gatherer suivant effectue un travail similaire à un Collectors.joining() via une opération séquentielle sur les éléments du flux.

Java

static Gatherer<CharSequence, ?, String> join( CharSequence separator ) {
  final class State {
    StringBuilder builder = null;
  }

  return Gatherer.ofSequential(
    State::new, // initializer
    Gatherer.Integrator.ofGreedy( (state, element, _) -> { // integrator
      if ( state.builder == null )
        state.builder = new StringBuilder( element );
      else
        state.builder.append( separator ).append( element );
      return true;
    }),
    (state, downstream) -> { // finisher
      if ( !downstream.isRejecting() && state.builder != null )
        downstream.push( state.builder.toString() );
    }
  );
}

Stream.of( "foo", "bar", "baz", "qux" )
  .gather( join(", ") )
  .forEach( System.out::println ); // Sortie : "foo, bar, baz, qux"

Ici, on maintient un état mutable State pour accumuler de manière efficace l’opération de join sur les éléments successifs. Notez que l’intégrateur ne publie pas dans son flux aval dans ce cas particulier et nous pouvons donc utiliser une optimisation via Integrator.ofGreedy() pour indiquer que l’on va consommer l’intégralité du flux. Seul l’opération finale va publier (si besoin).

💡On notera l’usage de ? dans la déclaration du type du Gatherer ici car ce dernier doit inclure le type de l’état mutable, or celui-ci n’est pas accessible en dehors de la fonction join.

Des Gatherers prêts à l’emploi

Le JDK 24 fournit plusieurs Gatherers prédéfinis via java.util.stream.Gatherers, notamment :

fold( initial, folder ) : agrégation incrémentale d’un état mutable.
scan( initial, scanner ) : accumulation incrémentale d’un état sur le flux.
windowFixed( windowSize ) : regroupement des éléments en fenêtres de taille fixe.
windowSliding( windowSize ) : fenêtres glissantes sur le flux.
mapConcurrent( maxConcurrency, mapper ) : application concurrente d’une transformation avec préservation de l’ordre.

A titre d’illustration, voici un exemple d’usage du fenêtrage de taille fixe :

Java

final var windows = Stream.of( 1, 2, 3, 4, 5, 6, 7, 8 )
  .gather( Gatherers.windowFixed(3) )
  .toList();
System.out.println( windows ); // Sortie : [[1, 2, 3], [4, 5, 6], [7, 8]]

Pourquoi adopter l’API Gatherer ?

L’API Gatherer vient combler un manque important dans l’API Stream, offrant une flexibilité inédite aux développeurs travaillant avec les flux de données. Bien que son apprentissage puisse demander un certain effort initial, elle permet de :

Exprimer des transformations complexes directement dans un pipeline Stream.
Simplifier et rendre plus lisible le code de traitement de données.
Optimiser les performances grâce à la gestion avancée de l’état et du parallélisme.

Si vous avez déjà été limité par les opérations intermédiaires standard des flux Java, il est temps d’explorer gather() et de repousser les frontières des transformations de données dans vos applications Java 24 !