Kopieren Sie einen Stream, um zu vermeiden, dass der Stream bereits bearbeitet oder geschlossen wurde.

Ich möchte einen Java 8-Stream duplizieren, damit ich zweimal damit umgehen kann. Ich kann collectals Liste und neue Streams daraus bekommen;

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

Aber ich denke, es sollte einen effizienteren / eleganteren Weg geben.

Gibt es eine Möglichkeit, den Stream zu kopieren, ohne ihn in eine Sammlung umzuwandeln?

Ich arbeite tatsächlich mit einem Stream von Eithers, möchte also die linke Projektion auf eine Weise verarbeiten, bevor ich auf die rechte Projektion übergehe und auf eine andere Weise damit umgehe. Ein bisschen wie dieses (mit dem ich bisher gezwungen bin, den toListTrick anzuwenden ).

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 -> ... );

Ich denke, Ihre Annahme über Effizienz ist etwas rückwärts. Sie erhalten diese enorme Amortisation, wenn Sie die Daten nur einmal verwenden, da Sie sie nicht speichern müssen. Streams bieten Ihnen leistungsstarke "Loop Fusion" -Optimierungen, mit denen Sie die gesamten Daten effizient durch die Pipeline fließen lassen können.

Wenn Sie dieselben Daten wiederverwenden möchten, müssen Sie sie per Definition entweder zweimal (deterministisch) generieren oder speichern. Wenn es sich bereits in einer Sammlung befindet, großartig; dann ist es billig, es zweimal zu wiederholen.

Wir haben im Design mit "gegabelten Streams" experimentiert. Was wir fanden, war, dass die Unterstützung echte Kosten hatte; es belastete den allgemeinen Fall (einmalige Verwendung) auf Kosten des ungewöhnlichen Falls. Das große Problem bestand darin, "was passiert, wenn die beiden Pipelines nicht mit der gleichen Rate Daten verbrauchen". Jetzt sind Sie sowieso wieder beim Puffern. Dies war ein Merkmal, das offensichtlich nicht das Gewicht hatte.

Wenn Sie dieselben Daten wiederholt bearbeiten möchten, speichern Sie sie entweder oder strukturieren Sie Ihre Vorgänge als Verbraucher und gehen Sie wie folgt vor:

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

Sie können sich auch die RxJava-Bibliothek ansehen, da sich das Verarbeitungsmodell besser für diese Art von "Stream Forking" eignet.

Sie können eine lokale Variable mit a verwenden Supplier, um allgemeine Teile der Stream-Pipeline einzurichten.

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

Streams wiederverwenden

Java 8-Streams können nicht wiederverwendet werden. Sobald Sie eine Terminaloperation aufrufen, wird der Stream geschlossen:

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)

Um diese Einschränkung zu überwinden, müssen wir für jede Terminaloperation, die wir ausführen möchten, eine neue Stream-Kette erstellen, z. B. könnten wir einen Stream-Lieferanten erstellen, um einen neuen Stream mit allen bereits eingerichteten Zwischenoperationen zu erstellen:

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

Jeder Aufruf zum get()Erstellen eines neuen Streams, in dem wir gespeichert sind, um die gewünschte Terminaloperation aufzurufen.

Verwenden Sie a Supplier, um den Stream für jede Beendigungsoperation zu erstellen.

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

Wenn Sie einen Stream dieser Sammlung benötigen streamSupplier.get(), können Sie einen neuen Stream abrufen.

Beispiele:

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

Wir haben eine duplicate()Methode für Streams in jOOλ implementiert , einer Open Source-Bibliothek, die wir erstellt haben, um die Integrationstests für jOOQ zu verbessern . Im Wesentlichen können Sie einfach schreiben:

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

Intern gibt es einen Puffer, in dem alle Werte gespeichert sind, die von einem Stream verbraucht wurden, aber nicht von dem anderen. Das ist wahrscheinlich so effizient wie es nur geht, wenn Ihre beiden Streams ungefähr gleich schnell verbraucht werden und wenn Sie mit dem Mangel an Thread-Sicherheit leben können .

So funktioniert der Algorithmus:

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()));
}

Mehr Quellcode hier

Tuple2ist wahrscheinlich wie Ihr Pairwährend Typ, Seqist Streammit einigen Verbesserungen.

Sie können beispielsweise einen Stream mit ausführbaren Dateien erstellen:

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

Wo failureund wo successsind die Operationen anzuwenden? Dies erstellt jedoch einige temporäre Objekte und ist möglicherweise nicht effizienter, als von einer Sammlung auszugehen und sie zweimal zu streamen / zu iterieren.

Eine andere Möglichkeit, die Elemente mehrmals zu behandeln, ist die Verwendung von Stream.peek (Consumer) :

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

peek(Consumer) kann so oft wie nötig verkettet werden.

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

cyclops-react , eine Bibliothek, zu der ich beitrage, verfügt über eine statische Methode, mit der Sie einen Stream duplizieren können (und ein jOOλ-Tupel von Streams zurückgibt).

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

Siehe Kommentare, es gibt Leistungseinbußen, die auftreten, wenn Duplikate in einem vorhandenen Stream verwendet werden. Eine leistungsfähigere Alternative wäre die Verwendung von Streamable:

Es gibt auch eine (faule) Streamable-Klasse, die aus einem Stream, Iterable oder Array erstellt und mehrmals wiedergegeben werden kann.

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

AsStreamable.synchronizedFromStream (Stream) - kann verwendet werden, um ein Streamable zu erstellen, das die Backing-Sammlung träge auf eine Weise auffüllt, die für mehrere Threads freigegeben werden kann. Streamable.fromStream (Stream) verursacht keinen Synchronisationsaufwand.

Für dieses spezielle Problem können Sie auch die Partitionierung verwenden. Etwas wie

     // 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)

Wir können Stream Builder zum Zeitpunkt des Lesens oder Iterierens eines Streams verwenden. Hier ist das Dokument von Stream Builder .

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

Anwendungsfall

Angenommen, wir haben einen Mitarbeiter-Stream und müssen diesen Stream verwenden, um Mitarbeiterdaten in eine Excel-Datei zu schreiben und dann die Mitarbeitersammlung / -tabelle zu aktualisieren. [Dies ist nur ein Anwendungsfall, um die Verwendung von Stream Builder zu zeigen]:

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();

Ich hatte ein ähnliches Problem und konnte mir drei verschiedene Zwischenstrukturen vorstellen, aus denen ich eine Kopie des Streams erstellen konnte: a List, ein Array und a Stream.Builder. Ich habe ein kleines Benchmark-Programm geschrieben, das darauf hinwies, dass es aus Sicht der Leistung Listetwa 30% langsamer war als die beiden anderen, die ziemlich ähnlich waren.

Der einzige Nachteil beim Konvertieren in ein Array besteht darin, dass es schwierig ist, wenn Ihr Elementtyp ein generischer Typ ist (was in meinem Fall der Fall war). deshalb bevorzuge ich a Stream.Builder.

Am Ende habe ich eine kleine Funktion geschrieben, die Folgendes erzeugt 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);
}

Ich kann dann eine Kopie jedes Streams erstellen, strindem ich str.collect(copyCollector())dies tue, was der idiomatischen Verwendung von Streams entspricht.