Copiez un flux pour éviter que "le flux a déjà été exploité ou fermé"

Je voudrais dupliquer un flux Java 8 pour pouvoir le gérer deux fois. Je peux collectcomme liste et obtenir de nouveaux flux à partir de cela;

// doSomething() returns a stream
List<A> thing = doSomething().collect(toList());
thing.stream()... // do stuff
thing.stream()... // do other stuff

Mais je pense qu'il devrait y avoir un moyen plus efficace / élégant.

Existe-t-il un moyen de copier le flux sans le transformer en collection?

Je travaille en fait avec un flux de Eithers, je veux donc traiter la projection de gauche dans un sens avant de passer à la projection de droite et de traiter cette autre manière. Un peu comme ça (avec lequel, jusqu'à présent, je suis obligé d'utiliser l' toListastuce).

List<Either<Pair<A, Throwable>, A>> results = doSomething().collect(toList());

Stream<Pair<A, Throwable>> failures = results.stream().flatMap(either -> either.left());
failures.forEach(failure -> ... );

Stream<A> successes = results.stream().flatMap(either -> either.right());
successes.forEach(success -> ... );

Je pense que votre hypothèse sur l'efficacité est un peu rétrograde. Vous obtenez ce gain d'efficacité énorme si vous n'utilisez les données qu'une seule fois, car vous n'avez pas à les stocker, et les flux vous offrent de puissantes optimisations de «fusion en boucle» qui vous permettent de faire circuler efficacement toutes les données dans le pipeline.

Si vous souhaitez réutiliser les mêmes données, vous devez par définition les générer deux fois (de manière déterministe) ou les stocker. S'il se trouve déjà dans une collection, tant mieux; puis l'itérer deux fois n'est pas cher.

Nous avons expérimenté dans la conception avec des «flux fourchus». Ce que nous avons constaté, c'est que soutenir cela avait des coûts réels; il alourdissait le cas courant (utilisation unique) aux dépens du cas rare. Le gros problème était de savoir «ce qui se passe lorsque les deux pipelines ne consomment pas de données au même rythme». De toute façon, vous revenez à la mise en mémoire tampon. C'était une caractéristique qui ne portait clairement pas son poids.

Si vous souhaitez utiliser les mêmes données à plusieurs reprises, stockez-les ou structurez vos opérations en consommateurs et procédez comme suit:

stream()...stuff....forEach(e -> { consumerA(e); consumerB(e); });

Vous pouvez également regarder dans la bibliothèque RxJava, car son modèle de traitement se prête mieux à ce type de "stream forking".

Vous pouvez utiliser une variable locale avec un Supplier pour configurer les parties communes du pipeline de flux.

De http://winterbe.com/posts/2014/07/31/java8-stream-tutorial-examples/ :

Réutilisation des flux

Les flux Java 8 ne peuvent pas être réutilisés. Dès que vous appelez une opération de terminal, le flux est fermé:

Stream<String> stream = Stream.of("d2", "a2", "b1", "b3", "c")
    .filter(s -> s.startsWith("a"));
stream.anyMatch(s -> true);    // ok
stream.noneMatch(s -> true);   // exception

Calling `noneMatch` after `anyMatch` on the same stream results in the following exception:
java.lang.IllegalStateException: stream has already been operated upon or closed
at 
java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229)
at 
java.util.stream.ReferencePipeline.noneMatch(ReferencePipeline.java:459)
at com.winterbe.java8.Streams5.test7(Streams5.java:38)
at com.winterbe.java8.Streams5.main(Streams5.java:28)

Pour surmonter cette limitation, nous devons créer une nouvelle chaîne de flux pour chaque opération de terminal que nous voulons exécuter, par exemple, nous pourrions créer un fournisseur de flux pour construire un nouveau flux avec toutes les opérations intermédiaires déjà configurées:

Supplier<Stream<String>> streamSupplier =
    () -> Stream.of("d2", "a2", "b1", "b3", "c")
            .filter(s -> s.startsWith("a"));

streamSupplier.get().anyMatch(s -> true);   // ok
streamSupplier.get().noneMatch(s -> true);  // ok

Chaque appel à get()construit un nouveau flux sur lequel nous sommes sauvegardés pour appeler l'opération de terminal souhaitée.

Utilisez a Supplierpour produire le flux pour chaque opération de terminaison.

Supplier<Stream<Integer>> streamSupplier = () -> list.stream();

Chaque fois que vous avez besoin d'un flux de cette collection, utilisez streamSupplier.get()pour obtenir un nouveau flux.

Exemples:

1. streamSupplier.get().anyMatch(predicate);
2. streamSupplier.get().allMatch(predicate2);

Nous avons implémenté une duplicate()méthode pour les flux dans jOOλ , une bibliothèque Open Source que nous avons créée pour améliorer les tests d'intégration pour jOOQ . Essentiellement, vous pouvez simplement écrire:

Tuple2<Seq<A>, Seq<A>> duplicates = Seq.seq(doSomething()).duplicate();

En interne, il existe un tampon stockant toutes les valeurs qui ont été consommées à partir d'un flux mais pas de l'autre. C'est probablement aussi efficace que si vos deux flux sont consommés à peu près au même rythme, et si vous pouvez vivre avec le manque de sécurité des threads .

Voici comment fonctionne l'algorithme:

static <T> Tuple2<Seq<T>, Seq<T>> duplicate(Stream<T> stream) {
    final List<T> gap = new LinkedList<>();
    final Iterator<T> it = stream.iterator();

    @SuppressWarnings("unchecked")
    final Iterator<T>[] ahead = new Iterator[] { null };

    class Duplicate implements Iterator<T> {
        @Override
        public boolean hasNext() {
            if (ahead[0] == null || ahead[0] == this)
                return it.hasNext();

            return !gap.isEmpty();
        }

        @Override
        public T next() {
            if (ahead[0] == null)
                ahead[0] = this;

            if (ahead[0] == this) {
                T value = it.next();
                gap.offer(value);
                return value;
            }

            return gap.poll();
        }
    }

    return tuple(seq(new Duplicate()), seq(new Duplicate()));
}

Plus de code source ici

Tuple2est probablement comme votre Pairtype alors Seqest Streamavec quelques améliorations.

Vous pouvez créer un flux d'exécutables (par exemple):

results.stream()
    .flatMap(either -> Stream.<Runnable> of(
            () -> failure(either.left()),
            () -> success(either.right())))
    .forEach(Runnable::run);

Où failureet successsont les opérations à appliquer. Cela créera cependant un certain nombre d'objets temporaires et ne sera peut-être pas plus efficace que de partir d'une collection et de la diffuser / itérer deux fois.

Une autre façon de gérer les éléments plusieurs fois consiste à utiliser Stream.peek (Consumer) :

doSomething().stream()
.peek(either -> handleFailure(either.left()))
.foreach(either -> handleSuccess(either.right()));

peek(Consumer) peut être enchaîné autant de fois que nécessaire.

doSomething().stream()
.peek(element -> handleFoo(element.foo()))
.peek(element -> handleBar(element.bar()))
.peek(element -> handleBaz(element.baz()))
.foreach(element-> handleQux(element.qux()));

cyclops-react , une bibliothèque à laquelle je contribue, a une méthode statique qui vous permettra de dupliquer un Stream (et renvoie un jOOλ Tuple of Streams).

    Stream<Integer> stream = Stream.of(1,2,3);
    Tuple2<Stream<Integer>,Stream<Integer>> streams =  StreamUtils.duplicate(stream);

Voir les commentaires, il y a une pénalité de performance qui sera encourue lors de l'utilisation de dupliquer sur un flux existant. Une alternative plus performante serait d'utiliser Streamable: -

Il existe également une classe Streamable (paresseuse) qui peut être construite à partir d'un Stream, Iterable ou Array et rejouée plusieurs fois.

    Streamable<Integer> streamable = Streamable.of(1,2,3);
    streamable.stream().forEach(System.out::println);
    streamable.stream().forEach(System.out::println);

AsStreamable.synchronizedFromStream (stream) - peut être utilisé pour créer un Streamable qui remplira paresseusement sa collection de sauvegarde, de manière à pouvoir être partagé entre les threads. Streamable.fromStream (stream) n'entraînera aucune surcharge de synchronisation.

Pour ce problème particulier, vous pouvez également utiliser le partitionnement. Quelque chose comme

     // Partition Eighters into left and right
     List<Either<Pair<A, Throwable>, A>> results = doSomething();
     Map<Boolean, Object> passingFailing = results.collect(Collectors.partitioningBy(s -> s.isLeft()));
     passingFailing.get(true) <- here will be all passing (left values)
     passingFailing.get(false) <- here will be all failing (right values)

Nous pouvons utiliser Stream Builder au moment de la lecture ou de l'itération d'un flux. Voici le document de Stream Builder .

https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.Builder.html

Cas d'utilisation

Disons que nous avons un flux d'employés et que nous devons utiliser ce flux pour écrire les données des employés dans un fichier Excel, puis mettre à jour la collection / table des employés [Ceci est juste un cas d'utilisation pour montrer l'utilisation de Stream Builder]:

Stream.Builder<Employee> builder = Stream.builder();

employee.forEach( emp -> {
   //store employee data to excel file 
   // and use the same object to build the stream.
   builder.add(emp);
});

//Now this stream can be used to update the employee collection
Stream<Employee> newStream = builder.build();

J'ai eu un problème similaire et je pouvais penser à trois structures intermédiaires différentes à partir desquelles créer une copie du flux: a List, un tableau et a Stream.Builder. J'ai écrit un petit programme de référence, qui suggérait que du point de vue de la performance, il Listétait environ 30% plus lent que les deux autres qui étaient assez similaires.

Le seul inconvénient de la conversion en tableau est qu'il est délicat si votre type d'élément est un type générique (ce qui dans mon cas c'était); donc je préfère utiliser un Stream.Builder.

J'ai fini par écrire une petite fonction qui crée un Collector:

private static <T> Collector<T, Stream.Builder<T>, Stream<T>> copyCollector()
{
    return Collector.of(Stream::builder, Stream.Builder::add, (b1, b2) -> {
        b2.build().forEach(b1);
        return b1;
    }, Stream.Builder::build);
}

Je peux ensuite faire une copie de n'importe quel flux stren faisant str.collect(copyCollector())ce qui me semble tout à fait conforme à l'utilisation idiomatique des flux.