Kartesisches Produkt von Streams in Java 8 als Stream (nur Streams)

Question

Ich möchte eine Methode erstellen, die einen Strom von Elementen erzeugt, die kartesische Produkte mehrerer gegebener Streams sind (am Ende durch eine Binärdatei zum selben Typ zusammengefasst) Operator). Bitte beachten Sie, dass beide Argumente und Ergebnisse Streams sind, nicht Sammlungen.

beispielsweise für zwei Ströme von {A, B} und {X, Y} Ich würde es Strom von Werten erzeugen mag {AX, AY, BX, BY} (einfache Verkettung verwendet wird, zum Aggregieren der Strings). Bisher habe ich kam mit diesem Code auf:

private static <T> Stream<T> cartesian(BinaryOperator<T> aggregator, Stream<T>... streams) { 
    Stream<T> result = null; 

    for (Stream<T> stream : streams) { 
     if (result == null) { 
      result = stream; 
     } else { 
      result = result.flatMap(m -> stream.map(n -> aggregator.apply(m, n))); 
     } 
    } 

    return result; 
} 

Dies ist mein Wunsch Anwendungsfall:

Stream<String> result = cartesian(
    (a, b) -> a + b, 
    Stream.of("A", "B"), 
    Stream.of("X", "Y") 
); 

System.out.println(result.collect(Collectors.toList())); 

Erwartetes Ergebnis: AX, AY, BX, BY.

Ein weiteres Beispiel:

Stream<String> result = cartesian(
    (a, b) -> a + b, 
    Stream.of("A", "B"), 
    Stream.of("K", "L"), 
    Stream.of("X", "Y") 
); 

Erwartetes Ergebnis: AKX, AKY, ALX, ALY, BKX, BKY, BLX, BLY.

Allerdings, wenn ich den Code ausführen, bekomme ich diesen Fehler:

Illegal: Strom bereits operiert worden ist oder geschlossen

Wo wird der Strom verbraucht? Durch flatMap? Kann es leicht repariert werden?

Answer 1

Vorbei die Ströme in Ihrem Beispiel ist nie besser als vorbei Listen:

private static <T> Stream<T> cartesian(BinaryOperator<T> aggregator, List<T>... lists) { 
    ... 
} 

Und es wie folgt verwenden:

Stream<String> result = cartesian(
    (a, b) -> a + b, 
    Arrays.asList("A", "B"), 
    Arrays.asList("K", "L"), 
    Arrays.asList("X", "Y") 
); 

In beiden Fällen erstellen Sie einen Imp legate Array von Varargs und benutze es als Datenquelle, so ist die Faulheit imaginär. Ihre Daten werden tatsächlich in den Arrays gespeichert.

In den meisten Fällen ist der resultierende kartesische Produktstrom viel länger als die Eingaben, daher gibt es praktisch keinen Grund, die Eingaben träge zu machen. Wenn Sie zum Beispiel fünf Listen mit fünf Elementen (insgesamt 25) haben, erhalten Sie den resultierenden Strom von 3125 Elementen. Es ist also kein großes Problem, 25 Elemente im Speicher zu speichern. Tatsächlich sind sie in den meisten praktischen Fällen bereits im Speicher gespeichert.

Um den Strom von kartesischen Produkten zu erzeugen, müssen Sie alle Streams (außer dem ersten) ständig "zurückspulen".Um zurückzuspulen, sollten die Streams in der Lage sein, die ursprünglichen Daten immer wieder neu zu laden, sie entweder zu puffern (was Ihnen nicht gefällt) oder sie wieder von der Quelle zu holen (Sammlung, Array, Datei, Netzwerk, Zufallszahlen, etc.)) und führe immer wieder alle Zwischenoperationen durch. Wenn die Quell- und Zwischenoperationen langsam sind, ist die Lösung möglicherweise langsamer als die Pufferlösung. Wenn Ihre Quelle die Daten nicht mehr erzeugen kann (z. B. Zufallszahlengenerator, der nicht die gleichen Zahlen erzeugen kann, die er zuvor erzeugt hat), ist Ihre Lösung falsch.

Trotzdem ist total faule Lösung möglich. Verwenden Sie einfach nicht Ströme, aber Strom Lieferanten:

private static <T> Stream<T> cartesian(BinaryOperator<T> aggregator, 
             Supplier<Stream<T>>... streams) { 
    return Arrays.stream(streams) 
     .reduce((s1, s2) -> 
      () -> s1.get().flatMap(t1 -> s2.get().map(t2 -> aggregator.apply(t1, t2)))) 
     .orElse(Stream::empty).get(); 
} 

Die Lösung ist interessant, wie wir den Strom von Lieferanten schaffen und reduzieren die resultierenden Lieferanten zu erhalten und es schließlich nennen. Verbrauch: [? Wie kann ich cartesianischen Produkt mit Java 8-Streams machen]

Stream<String> result = cartesian(
      (a, b) -> a + b, 
     () -> Stream.of("A", "B"), 
     () -> Stream.of("K", "L"), 
     () -> Stream.of("X", "Y") 
     ); 
result.forEach(System.out::println); 

Answer 2

stream wird in der zweiten Iteration in der Operation flatMap verbraucht. Du musst also jedes Mal einen neuen Stream erstellen, wenn du dein Ergebnis map hast. Daher müssen Sie die stream im Voraus sammeln, um in jeder Iteration einen neuen Stream zu erhalten.

private static <T> Stream<T> cartesian(BiFunction<T, T, T> aggregator, Stream<T>... streams) { 
    Stream<T> result = null; 
    for (Stream<T> stream : streams) { 
     if (result == null) { 
      result = stream; 
     } else { 
      Collection<T> s = stream.collect(Collectors.toList()); 
      result = result.flatMap(m -> s.stream().map(n -> aggregator.apply(m, n))); 
     } 
    } 
    return result; 
} 

Oder noch kürzer:

private static <T> Stream<T> cartesian(BiFunction<T, T, T> aggregator, Stream<T>... streams) { 
    return Arrays.stream(streams).reduce((r, s) -> { 
     List<T> collect = s.collect(Collectors.toList()); 
     return r.flatMap(m -> collect.stream().map(n -> aggregator.apply(m, n))); 
    }).orElse(Stream.empty()); 
}