Можете ли вы разделить поток на два потока?

Ответ 1

Не совсем. Вы не можете получить два Stream из одного; это не имеет смысла - как бы вы перебирали один, не создавая при этом другого? Поток может работать только один раз.

Однако, если вы хотите сбросить их в список или что-то еще, вы можете сделать

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

Ответ 2

Для этого может использоваться сборщик.

• Для двух категорий используйте Collectors.partitioningBy() factory.

Это создаст Map от Boolean до List и поместит элементы в один или другой список на основе Predicate.

Примечание. Поскольку поток нужно потреблять целиком, это не может работать на бесконечных потоках. Поскольку поток все равно потребляется, этот метод просто помещает их в списки вместо создания нового потока с памятью.

Кроме того, нет необходимости в итераторе, даже не в примере с заголовками, которые вы указали.

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

• Для получения дополнительных категорий используйте 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());

Если потоки не Stream, но один из примитивных потоков, таких как IntStream, то этот метод .collect(Collectors) недоступен. Вам придется сделать это вручную без коллектора factory. Это выполнение выглядит следующим образом:

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

Edit

Как указано, вышеописанный "обходной путь" не является потокобезопасным. Переход к нормальному Stream перед сборкой - это путь:

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

Ответ 3

К сожалению, то, о чем вы просите, напрямую недооценивается в JavaDoc потока:

Поток должен работать (при вызове промежуточного или терминального поток) только один раз. Это исключает, например, "раздвоение", потоки, где один и тот же источник подает два или более конвейера или множественные обходы одного и того же потока.

Вы можете обойти это с помощью peek или других методов, если вы действительно желаете такого типа поведения. В этом случае вы должны сделать вместо того, чтобы пытаться вернуть два потока из того же исходного источника потока с помощью фильтра forking, вы должны дублировать поток и соответствующим образом фильтровать каждый из дубликатов.

Однако вы можете пересмотреть, если Stream является подходящей структурой для вашего использования.

Ответ 4

Я наткнулся на этот вопрос на себя, и я чувствую, что разветвленный поток имеет некоторые варианты использования, которые могут оказаться действительными. Я написал код ниже в качестве потребителя, чтобы он ничего не делал, но вы могли применить его к функциям и всему, что вы могли бы встретить.

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

Теперь реализация кода может быть примерно такой:

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

Ответ 5

Это противоречит общему механизму Stream. Скажем, вы можете разделить Stream S0 на Sa и Sb, как вы хотели. Выполняя любую операцию терминала, скажем count(), на Sa обязательно "потребляют" все элементы в S0. Поэтому Sb потерял свой источник данных.

Раньше Stream имел метод tee(), который, по-моему, дублировал поток на два. Теперь он удален.

У потока есть метод peek(), хотя вы можете использовать его для достижения ваших требований.

Ответ 6

не совсем, но вы можете выполнить то, что вам нужно, вызывая Collectors.groupingBy(). вы создаете новую коллекцию и затем можете создавать потоки в этой новой коллекции.

Ответ 7

Это был самый плохой ответ, который я мог придумать.

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

}

Это берет поток целых чисел и разбивает их на 5. Для тех, кто больше 5, он фильтрует только четные числа и помещает их в список. В остальном он соединяет их с |.

выходы:

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

Его не идеально, поскольку он собирает все в промежуточные коллекции, разбивающие поток (и имеет слишком много аргументов!)

Ответ 8

Я наткнулся на этот вопрос, ища способ фильтровать определенные элементы из потока и регистрировать их как ошибки. Поэтому мне не нужно было разделить поток так же, как приложить преждевременное завершающее действие к предикату с ненавязчивым синтаксисом. Вот что я придумал:

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

Ответ 9

Как насчет:

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