Können Sie einen Stream in zwei Streams aufteilen?

Ich habe einen Datensatz, der durch einen Java 8-Stream dargestellt wird:

Stream<T> stream = ...;

Ich kann sehen, wie man es filtert, um eine zufällige Teilmenge zu erhalten - zum Beispiel

Random r = new Random();
PrimitiveIterator.OfInt coin = r.ints(0, 2).iterator();   
Stream<T> heads = stream.filter((x) -> (coin.nextInt() == 0));

Ich kann auch sehen, wie ich diesen Stream reduzieren kann, um beispielsweise zwei Listen zu erhalten, die zwei zufällige Hälften des Datensatzes darstellen, und diese dann wieder in Streams umzuwandeln. Aber gibt es eine direkte Möglichkeit, zwei Streams aus dem ersten zu generieren? Etwas wie

(heads, tails) = stream.[some kind of split based on filter]

Vielen Dank für jeden Einblick.

Nicht genau. Sie können nicht zwei Streamaus einem herausholen; Das macht keinen Sinn - wie würden Sie über eines iterieren, ohne das andere gleichzeitig generieren zu müssen? Ein Stream kann nur einmal bearbeitet werden.

Wenn Sie sie jedoch in eine Liste oder etwas anderes kopieren möchten, können Sie dies tun

stream.forEach((x) -> ((x == 0) ? heads : tails).add(x));

Hierfür kann ein Kollektor verwendet werden.

• Verwenden Sie für zwei Kategorien Collectors.partitioningBy()Factory.

Dadurch wird ein MapVon Booleanbis erstellt Listund Elemente basierend auf a in die eine oder andere Liste eingefügt Predicate.

Hinweis: Da der Stream als Ganzes verbraucht werden muss, kann dies bei unendlichen Streams nicht funktionieren. Und weil der Stream sowieso verbraucht wird, werden sie bei dieser Methode einfach in Listen eingefügt, anstatt einen neuen Stream mit Speicher zu erstellen. Sie können diese Listen jederzeit streamen, wenn Sie Streams als Ausgabe benötigen.

Außerdem ist der Iterator nicht erforderlich, auch nicht in dem von Ihnen bereitgestellten Nur-Kopf-Beispiel.

• Die binäre Aufteilung sieht folgendermaßen aus:

Random r = new Random();

Map<Boolean, List<String>> groups = stream
    .collect(Collectors.partitioningBy(x -> r.nextBoolean()));

System.out.println(groups.get(false).size());
System.out.println(groups.get(true).size());

• Verwenden Sie für weitere Kategorien eine Collectors.groupingBy()Factory.

Map<Object, List<String>> groups = stream
    .collect(Collectors.groupingBy(x -> r.nextInt(3)));
System.out.println(groups.get(0).size());
System.out.println(groups.get(1).size());
System.out.println(groups.get(2).size());

Falls die Streams nicht sind Stream, aber einer der primitiven Streams wie IntStream, dann ist diese .collect(Collectors)Methode nicht verfügbar. Sie müssen dies manuell ohne Sammlerfabrik tun. Die Implementierung sieht folgendermaßen aus:

[Beispiel 2.0 seit 2020-04-16]

    IntStream    intStream = IntStream.iterate(0, i -> i + 1).limit(100000).parallel();
    IntPredicate predicate = ignored -> r.nextBoolean();

    Map<Boolean, List<Integer>> groups = intStream.collect(
            () -> Map.of(false, new ArrayList<>(100000),
                         true , new ArrayList<>(100000)),
            (map, value) -> map.get(predicate.test(value)).add(value),
            (map1, map2) -> {
                map1.get(false).addAll(map2.get(false));
                map1.get(true ).addAll(map2.get(true ));
            });

In diesem Beispiel initialisiere ich die ArrayLists mit der vollen Größe der ursprünglichen Sammlung (falls dies überhaupt bekannt ist). Dies verhindert Größenänderungsereignisse auch im schlimmsten Fall, kann jedoch möglicherweise 2 * N * T-Speicherplatz verschlingen (N = anfängliche Anzahl von Elementen, T = Anzahl von Threads). Um den Platz gegen die Geschwindigkeit auszutauschen, können Sie ihn weglassen oder Ihre am besten fundierte Vermutung verwenden, z. B. die erwartete höchste Anzahl von Elementen in einer Partition (normalerweise etwas mehr als N / 2 für eine ausgeglichene Aufteilung).

Ich hoffe, ich beleidige niemanden mit einer Java 9-Methode. Informationen zur Java 8-Version finden Sie im Bearbeitungsverlauf.

Ich bin über diese Frage gestolpert und habe das Gefühl, dass ein gegabelter Stream einige Anwendungsfälle hat, die sich als gültig erweisen könnten. Ich habe den folgenden Code als Verbraucher geschrieben, damit er nichts tut, aber Sie können ihn auf Funktionen und alles andere anwenden, auf das Sie stoßen könnten.

class PredicateSplitterConsumer<T> implements Consumer<T>
{
  private Predicate<T> predicate;
  private Consumer<T>  positiveConsumer;
  private Consumer<T>  negativeConsumer;

  public PredicateSplitterConsumer(Predicate<T> predicate, Consumer<T> positive, Consumer<T> negative)
  {
    this.predicate = predicate;
    this.positiveConsumer = positive;
    this.negativeConsumer = negative;
  }

  @Override
  public void accept(T t)
  {
    if (predicate.test(t))
    {
      positiveConsumer.accept(t);
    }
    else
    {
      negativeConsumer.accept(t);
    }
  }
}

Jetzt könnte Ihre Code-Implementierung ungefähr so ​​aussehen:

personsArray.forEach(
        new PredicateSplitterConsumer<>(
            person -> person.getDateOfBirth().isPresent(),
            person -> System.out.println(person.getName()),
            person -> System.out.println(person.getName() + " does not have Date of birth")));

Leider ist das, wonach Sie fragen, im JavaDoc von Stream direkt verpönt :

Ein Stream sollte nur einmal bearbeitet werden (Aufruf einer Zwischen- oder Terminal-Stream-Operation). Dies schließt beispielsweise "gegabelte" Streams aus, bei denen dieselbe Quelle zwei oder mehr Pipelines oder mehrere Durchläufe desselben Streams speist.

Sie können dies mit peekoder mit anderen Methoden umgehen, wenn Sie diese Art von Verhalten wirklich wünschen. In diesem Fall sollten Sie nicht versuchen, zwei Streams von derselben ursprünglichen Stream-Quelle mit einem Gabelfilter zu sichern, sondern Ihren Stream duplizieren und jedes der Duplikate entsprechend filtern.

Möglicherweise möchten Sie jedoch erneut prüfen, ob a Streamdie geeignete Struktur für Ihren Anwendungsfall ist.

Dies ist gegen den allgemeinen Mechanismus von Stream. Angenommen, Sie können Stream S0 wie gewünscht in Sa und Sb aufteilen. Wenn Sie beispielsweise eine Terminaloperation count()an Sa ausführen, werden alle Elemente in S0 "verbraucht". Daher hat Sb seine Datenquelle verloren.

Zuvor hatte Stream eine tee() Methode, die einen Stream in zwei dupliziert. Es ist jetzt entfernt.

Stream verfügt jedoch über eine peek () -Methode, mit der Sie möglicherweise Ihre Anforderungen erfüllen können.

nicht genau, aber Sie können möglicherweise erreichen, was Sie benötigen, indem Sie aufrufen Collectors.groupingBy(). Sie erstellen eine neue Sammlung und können dann Streams für diese neue Sammlung instanziieren.

Dies war die am wenigsten schlechte Antwort, die ich finden konnte.

import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;

public class Test {

    public static <T, L, R> Pair<L, R> splitStream(Stream<T> inputStream, Predicate<T> predicate,
            Function<Stream<T>, L> trueStreamProcessor, Function<Stream<T>, R> falseStreamProcessor) {

        Map<Boolean, List<T>> partitioned = inputStream.collect(Collectors.partitioningBy(predicate));
        L trueResult = trueStreamProcessor.apply(partitioned.get(Boolean.TRUE).stream());
        R falseResult = falseStreamProcessor.apply(partitioned.get(Boolean.FALSE).stream());

        return new ImmutablePair<L, R>(trueResult, falseResult);
    }

    public static void main(String[] args) {

        Stream<Integer> stream = Stream.iterate(0, n -> n + 1).limit(10);

        Pair<List<Integer>, String> results = splitStream(stream,
                n -> n > 5,
                s -> s.filter(n -> n % 2 == 0).collect(Collectors.toList()),
                s -> s.map(n -> n.toString()).collect(Collectors.joining("|")));

        System.out.println(results);
    }

}

Dies nimmt einen Strom von ganzen Zahlen und teilt sie bei 5 auf. Bei mehr als 5 werden nur gerade Zahlen gefiltert und in eine Liste aufgenommen. Für den Rest verbindet es sie mit |.

Ausgänge:

 ([6, 8],0|1|2|3|4|5)

Es ist nicht ideal, da es alles in Zwischensammlungen sammelt, die den Strom brechen (und zu viele Argumente hat!).

Ich bin auf diese Frage gestoßen, als ich nach einer Möglichkeit gesucht habe, bestimmte Elemente aus einem Stream herauszufiltern und als Fehler zu protokollieren. Ich musste den Stream also nicht wirklich aufteilen, sondern einem Prädikat mit unauffälliger Syntax eine vorzeitige Beendigungsaktion hinzufügen. Folgendes habe ich mir ausgedacht:

public class MyProcess {
    /* Return a Predicate that performs a bail-out action on non-matching items. */
    private static <T> Predicate<T> withAltAction(Predicate<T> pred, Consumer<T> altAction) {
    return x -> {
        if (pred.test(x)) {
            return true;
        }
        altAction.accept(x);
        return false;
    };

    /* Example usage in non-trivial pipeline */
    public void processItems(Stream<Item> stream) {
        stream.filter(Objects::nonNull)
              .peek(this::logItem)
              .map(Item::getSubItems)
              .filter(withAltAction(SubItem::isValid,
                                    i -> logError(i, "Invalid")))
              .peek(this::logSubItem)
              .filter(withAltAction(i -> i.size() > 10,
                                    i -> logError(i, "Too large")))
              .map(SubItem::toDisplayItem)
              .forEach(this::display);
    }
}

Kürzere Version, die Lombok verwendet

import java.util.function.Consumer;
import java.util.function.Predicate;

import lombok.RequiredArgsConstructor;

/**
 * Forks a Stream using a Predicate into postive and negative outcomes.
 */
@RequiredArgsConstructor
@FieldDefaults(makeFinal = true, level = AccessLevel.PROTECTED)
public class StreamForkerUtil<T> implements Consumer<T> {
    Predicate<T> predicate;
    Consumer<T> positiveConsumer;
    Consumer<T> negativeConsumer;

    @Override
    public void accept(T t) {
        (predicate.test(t) ? positiveConsumer : negativeConsumer).accept(t);
    }
}

Wie wäre es mit:

Supplier<Stream<Integer>> randomIntsStreamSupplier =
    () -> (new Random()).ints(0, 2).boxed();

Stream<Integer> tails =
    randomIntsStreamSupplier.get().filter(x->x.equals(0));
Stream<Integer> heads =
    randomIntsStreamSupplier.get().filter(x->x.equals(1));