¿Se puede dividir una secuencia en dos secuencias?

Tengo un conjunto de datos representado por una secuencia Java 8:

Stream<T> stream = ...;

Puedo ver cómo filtrarlo para obtener un subconjunto aleatorio, por ejemplo

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

También puedo ver cómo podría reducir esta secuencia para obtener, por ejemplo, dos listas que representan dos mitades aleatorias del conjunto de datos, y luego convertirlas nuevamente en secuencias. Pero, ¿hay una forma directa de generar dos secuencias a partir de la inicial? Algo como

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

Gracias por cualquier idea.

No exactamente. No puedes obtener dos Streams de uno; esto no tiene sentido: ¿cómo iteraría sobre uno sin necesidad de generar el otro al mismo tiempo? Una secuencia solo se puede operar una vez.

Sin embargo, si desea volcarlos en una lista o algo, podría hacer

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

Se puede usar un colector para esto.

• Para dos categorías, use la Collectors.partitioningBy()fábrica.

Esto creará un Mapdesde Booleanhasta Listy colocará elementos en una u otra lista según un Predicate.

Nota: Dado que la transmisión debe consumirse completa, esto no puede funcionar en transmisiones infinitas. Y debido a que el flujo se consume de todos modos, este método simplemente los coloca en listas en lugar de crear un nuevo flujo con memoria. Siempre puede transmitir esas listas si necesita transmisiones como salida.

Además, no es necesario el iterador, ni siquiera en el ejemplo de solo cabezas que proporcionó.

• La división binaria se ve así:

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

• Para más categorías, use una Collectors.groupingBy()fábrica.

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

En caso de que los flujos no lo sean Stream, pero uno de los flujos primitivos parezca IntStream, entonces este .collect(Collectors)método no está disponible. Tendrás que hacerlo de forma manual sin una fábrica de colectores. Su implementación se ve así:

[Ejemplo 2.0 desde 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 ));
            });

En este ejemplo, inicializo las ArrayLists con el tamaño completo de la colección inicial (si esto se conoce). Esto evita eventos de cambio de tamaño incluso en el peor de los casos, pero potencialmente puede engullir 2 * N * T espacio (N = número inicial de elementos, T = número de subprocesos). Para cambiar el espacio por la velocidad, puede omitirlo o usar su conjetura mejor educada, como el mayor número esperado de elementos en una partición (generalmente un poco más de N / 2 para una división equilibrada).

Espero no ofender a nadie usando un método Java 9. Para la versión Java 8, mire el historial de edición.

Me topé con esta pregunta y siento que una transmisión bifurcada tiene algunos casos de uso que podrían ser válidos. Escribí el siguiente código como consumidor para que no haga nada, pero podría aplicarlo a las funciones y cualquier otra cosa que pueda encontrar.

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

Ahora su implementación de código podría ser algo como esto:

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

Desafortunadamente, lo que pides está directamente mal visto en JavaDoc de Stream :

Una secuencia debe operarse (invocando una operación de secuencia intermedia o terminal) solo una vez. Esto excluye, por ejemplo, las secuencias "bifurcadas", donde la misma fuente alimenta dos o más canalizaciones, o múltiples recorridos de la misma secuencia.

Puede evitar esto utilizando peeku otros métodos si realmente desea ese tipo de comportamiento. En este caso, lo que debe hacer es, en lugar de intentar hacer una copia de seguridad de dos secuencias de la misma fuente de secuencia original con un filtro de bifurcación, duplicaría su secuencia y filtraría cada uno de los duplicados adecuadamente.

Sin embargo, es posible que desee reconsiderar si Streames la estructura adecuada para su caso de uso.

Esto va en contra del mecanismo general de Stream. Digamos que puede dividir Stream S0 a Sa y Sb como quisiera. Realizar cualquier operación de terminal, por ejemplo count(), en Sa necesariamente "consumirá" todos los elementos en S0. Por lo tanto, Sb perdió su fuente de datos.

Anteriormente, Stream tenía un tee()método, creo, que duplicaba un flujo a dos. Se ha eliminado ahora.

Sin embargo, Stream tiene un método peek (), es posible que pueda usarlo para cumplir con sus requisitos.

no exactamente, pero puede lograr lo que necesita invocando Collectors.groupingBy(). crea una nueva colección y luego puede instanciar transmisiones en esa nueva colección.

Esta fue la respuesta menos mala que se me ocurrió.

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

}

Esto toma una secuencia de enteros y los divide en 5. Para los mayores de 5, filtra solo los números pares y los coloca en una lista. Por lo demás, los une con |.

salidas:

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

No es ideal, ya que recopila todo en colecciones intermedias que rompen la corriente (¡y tiene demasiados argumentos!)

Me topé con esta pregunta mientras buscaba una forma de filtrar ciertos elementos de una secuencia y registrarlos como errores. Por lo tanto, realmente no necesitaba dividir la secuencia, sino adjuntar una acción de terminación prematura a un predicado con sintaxis discreta. Esto es lo que se me ocurrió:

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

Versión más corta que usa Lombok

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

Qué tal si:

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