Pouvez-vous diviser un flux de données en deux volets?

Un collecteur peut être utilisé pour cela.

• Pour les deux catégories, l'utilisation Collectors.partitioningBy() usine.

Cela va créer une Map de Boolean à List, et de mettre des articles dans l'une ou l'autre liste basée sur un Predicate.

Remarque: Depuis le flux doit être consommé entier, cela ne peut pas travailler sur les flux infinis. Parce que le flux est consommé de toute façon, cette méthode met tout simplement dans les Listes au lieu de faire un nouveau flux de données avec la mémoire.

Aussi, pas besoin de l'itérateur, pas même dans les chefs-seul exemple que vous avez fournis.

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

• Pour plus de catégories, utilisez un Collectors.groupingBy() usine.

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

Dans le cas où les flux ne sont Stream, mais l'une des primitives des cours d'eau,IntStream, alors ce .collect(Collectors) méthode n'est pas disponible. Vous aurez à le faire de la manière manuelle sans un collecteur d'usine. C'est la mise en œuvre ressemble à ceci:

IntStream intStream = IntStream.iterate(0, i -> i + 1).limit(1000000);

Predicate<Integer> p = x -> r.nextBoolean();
Map<Boolean, List<Integer>> groups = intStream.collect(() -> {
    Map<Boolean, List<Integer>> map = new HashMap<>();
    map.put(false, new ArrayList<>());
    map.put(true, new ArrayList<>());
    return map;
}, (map, x) -> {
    boolean partition = p.test(x);
    List<Integer> list = map.get(partition);
    list.add(x);
}, (map1, map2) -> {
    map1.get(false).addAll(map2.get(false));
    map1.get(true).addAll(map2.get(true));
});

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

Modifier

Comme l'a fait remarquer, au dessus de "solution" n'est pas thread-safe. La conversion à une vie normale Stream avant la collecte est le chemin à parcourir:

Stream<Integer> stream = intStream.boxed();

Je suis tombé sur cette question à mon auto et j'ai l'impression que fourchue flux a certains cas d'utilisation, qui pourrait s'avérer valide. J'ai écrit le code ci-dessous en tant que consommateur, de sorte qu'il ne fait rien mais on pourrait l'appliquer à des fonctions et à rien d'autre, vous pourriez trouver.

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

Maintenant votre code de mise en œuvre pourrait être quelque chose comme ceci:

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

pas exactement, mais vous pouvez être en mesure d'accomplir ce que vous avez besoin en invoquant Collectors.groupingBy(). vous créez une nouvelle Collection, et peut ensuite instancier streams sur cette nouvelle collection.

Je suis tombé sur cette question tout en cherchant un moyen de filtrer certains éléments d'un flux de données et les enregistre comme des erreurs. Je n'ai donc pas vraiment besoin de diviser le flux de façon plus joindre un prématuré de résiliation de l'action d'un prédicat avec discrète de la syntaxe. C'est ce que je suis venu avec:

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