Você pode dividir um fluxo em dois fluxos?

Eu tenho um conjunto de dados representado por um fluxo Java 8:

Stream<T> stream = ...;

Eu posso ver como filtrá-lo para obter um subconjunto aleatório - por exemplo

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

Também posso ver como reduzir esse fluxo para obter, por exemplo, duas listas representando duas metades aleatórias do conjunto de dados e depois transformá-las novamente em fluxos. Mas, existe uma maneira direta de gerar dois fluxos a partir do inicial? Algo como

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

Obrigado por qualquer insight.

Não exatamente. Você não pode obter dois Streams de um; isso não faz sentido - como você iteraria uma sem precisar gerar a outra ao mesmo tempo? Um fluxo pode ser operado apenas uma vez.

No entanto, se você deseja despejá-los em uma lista ou algo assim, você pode fazer

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

Um coletor pode ser usado para isso.

• Para duas categorias, use Collectors.partitioningBy()factory.

Isso criará um Mapde Booleanpara Liste colocará itens em uma ou outra lista com base em a Predicate.

Nota: Como o fluxo precisa ser consumido inteiro, isso não pode funcionar em fluxos infinitos. E como o fluxo é consumido de qualquer maneira, esse método simplesmente os coloca em Listas, em vez de criar um novo fluxo com memória. Você sempre pode transmitir essas listas se precisar de fluxos como saída.

Além disso, não há necessidade do iterador, nem mesmo no exemplo somente de cabeçote que você forneceu.

• A divisão binária é assim:

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 mais categorias, use uma 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());

Caso os fluxos não sejam Stream, mas um dos fluxos primitivos IntStream, esse .collect(Collectors)método não estará disponível. Você terá que fazer isso da maneira manual sem uma fábrica de colecionadores. Sua implementação é assim:

[Exemplo 2.0 desde 16/04 2020]

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

Neste exemplo, inicializo o ArrayLists com o tamanho total da coleção inicial (se é que isso é conhecido). Isso evita eventos de redimensionamento, mesmo no pior cenário, mas pode potencialmente devorar 2 * N * T de espaço (N = número inicial de elementos, T = número de encadeamentos). Para trocar o espaço pela velocidade, você pode deixá-lo de fora ou usar seu palpite melhor, como o maior número esperado de elementos em uma partição (normalmente um pouco acima de N / 2 para uma divisão equilibrada).

Espero não ofender ninguém usando o método Java 9. Para a versão do Java 8, consulte o histórico de edições.

Eu me deparei com essa pergunta e sinto que um fluxo bifurcado tem alguns casos de uso que podem ser válidos. Escrevi o código abaixo como consumidor para que ele não faça nada, mas você pode aplicá-lo a funções e qualquer outra coisa que possa 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);
    }
  }
}

Agora sua implementação de código pode ser algo como isto:

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

Infelizmente, o que você pede é desaprovado diretamente no JavaDoc of Stream :

Um fluxo deve ser operado (invocando uma operação de fluxo intermediária ou terminal) apenas uma vez. Isso exclui, por exemplo, fluxos "bifurcados", nos quais a mesma fonte alimenta dois ou mais pipelines, ou vários percursos do mesmo fluxo.

Você pode solucionar esse problema usando peekoutros métodos, se realmente desejar esse tipo de comportamento. Nesse caso, o que você deve fazer é, em vez de tentar fazer backup de dois fluxos da mesma fonte original do fluxo com um filtro de bifurcação, você duplicaria o fluxo e filtraria cada uma das duplicatas adequadamente.

No entanto, convém reconsiderar se a Streamé a estrutura apropriada para seu caso de uso.

Isso é contra o mecanismo geral do Stream. Digamos que você possa dividir o Stream S0 em Sa e Sb como desejar. Executar qualquer operação do terminal, digamos count(), no Sa, necessariamente "consumirá" todos os elementos em S0. Portanto, o Sb perdeu sua fonte de dados.

Anteriormente, tee()eu acho que o Stream tinha um método que duplicava um stream para dois. Foi removido agora.

O Stream possui um método peek (); talvez você possa usá-lo para atender aos seus requisitos.

não exatamente, mas você pode conseguir o que precisa invocando Collectors.groupingBy(). você cria uma nova coleção e pode instanciar fluxos nessa nova coleção.

Essa foi a resposta menos ruim que eu consegui encontrar.

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

}

Isso pega um fluxo de números inteiros e os divide em 5. Para aqueles com mais de 5, ele filtra apenas números pares e os coloca em uma lista. De resto, junta-os a |.

saídas:

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

Não é o ideal, pois reúne tudo em coleções intermediárias que quebram o fluxo (e tem muitos argumentos!)

Eu me deparei com essa pergunta enquanto procurava uma maneira de filtrar certos elementos de um fluxo e registrá-los como erros. Portanto, eu realmente não precisava dividir o fluxo, mas anexar uma ação de encerramento prematura a um predicado com sintaxe discreta. Isto é o que eu vim com:

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

Versão mais curta 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);
    }
}

E se:

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