Скопируйте поток, чтобы избежать «поток уже обработан или закрыт»

Я хотел бы продублировать поток Java 8, чтобы иметь возможность работать с ним дважды. Я могу collectкак список и получать новые потоки из этого;

// doSomething() returns a stream
List<A> thing = doSomething().collect(toList());
thing.stream()... // do stuff
thing.stream()... // do other stuff

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

Есть ли способ скопировать поток, не превращая его в коллекцию?

На самом деле я работаю с потоком Eithers, поэтому хочу обработать левую проекцию одним способом, прежде чем перейти к правой проекции и работать с ней по-другому. Что-то вроде этого (с которым я пока вынужден использовать этот toListтрюк).

List<Either<Pair<A, Throwable>, A>> results = doSomething().collect(toList());

Stream<Pair<A, Throwable>> failures = results.stream().flatMap(either -> either.left());
failures.forEach(failure -> ... );

Stream<A> successes = results.stream().flatMap(either -> either.right());
successes.forEach(success -> ... );

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

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

Мы экспериментировали в дизайне с «разветвленными потоками». Мы обнаружили, что поддержка этого требует реальных затрат; это обременяет общий случай (использовать один раз) за счет необычного случая. Большая проблема заключалась в том, «что происходит, когда два конвейера не потребляют данные с одинаковой скоростью». Теперь вы все равно вернетесь к буферизации. Это была особенность, которая явно не имела веса.

Если вы хотите работать с одними и теми же данными несколько раз, либо сохраните их, либо структурируйте свои операции как Потребители и выполните следующие действия:

stream()...stuff....forEach(e -> { consumerA(e); consumerB(e); });

Вы также можете изучить библиотеку RxJava, поскольку ее модель обработки лучше подходит для такого рода «разветвления потока».

Вы можете использовать локальную переменную с, Supplierчтобы настроить общие части конвейера потока.

С http://winterbe.com/posts/2014/07/31/java8-stream-tutorial-examples/ :

Повторное использование потоков

Потоки Java 8 нельзя использовать повторно. Как только вы вызываете любую операцию терминала, поток закрывается:

Stream<String> stream = Stream.of("d2", "a2", "b1", "b3", "c")
    .filter(s -> s.startsWith("a"));
stream.anyMatch(s -> true);    // ok
stream.noneMatch(s -> true);   // exception

Calling `noneMatch` after `anyMatch` on the same stream results in the following exception:
java.lang.IllegalStateException: stream has already been operated upon or closed
at 
java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229)
at 
java.util.stream.ReferencePipeline.noneMatch(ReferencePipeline.java:459)
at com.winterbe.java8.Streams5.test7(Streams5.java:38)
at com.winterbe.java8.Streams5.main(Streams5.java:28)

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

Supplier<Stream<String>> streamSupplier =
    () -> Stream.of("d2", "a2", "b1", "b3", "c")
            .filter(s -> s.startsWith("a"));

streamSupplier.get().anyMatch(s -> true);   // ok
streamSupplier.get().noneMatch(s -> true);  // ok

Каждый вызов to get()создает новый поток, в котором мы сохраняемся, чтобы вызвать желаемую операцию терминала.

Используйте a Supplierдля создания потока для каждой операции завершения.

Supplier<Stream<Integer>> streamSupplier = () -> list.stream();

Когда вам понадобится поток этой коллекции, используйте, streamSupplier.get()чтобы получить новый поток.

Примеры:

1. streamSupplier.get().anyMatch(predicate);
2. streamSupplier.get().allMatch(predicate2);

Мы реализовали duplicate()метод для потоков в jOOλ , библиотеке с открытым исходным кодом, которую мы создали для улучшения интеграционного тестирования для jOOQ . По сути, вы можете просто написать:

Tuple2<Seq<A>, Seq<A>> duplicates = Seq.seq(doSomething()).duplicate();

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

Вот как работает алгоритм:

static <T> Tuple2<Seq<T>, Seq<T>> duplicate(Stream<T> stream) {
    final List<T> gap = new LinkedList<>();
    final Iterator<T> it = stream.iterator();

    @SuppressWarnings("unchecked")
    final Iterator<T>[] ahead = new Iterator[] { null };

    class Duplicate implements Iterator<T> {
        @Override
        public boolean hasNext() {
            if (ahead[0] == null || ahead[0] == this)
                return it.hasNext();

            return !gap.isEmpty();
        }

        @Override
        public T next() {
            if (ahead[0] == null)
                ahead[0] = this;

            if (ahead[0] == this) {
                T value = it.next();
                gap.offer(value);
                return value;
            }

            return gap.poll();
        }
    }

    return tuple(seq(new Duplicate()), seq(new Duplicate()));
}

Больше исходного кода здесь

Tuple2вероятно, как ваш Pairтип, Seqно Streamс некоторыми улучшениями.

Вы можете создать поток исполняемых файлов (например):

results.stream()
    .flatMap(either -> Stream.<Runnable> of(
            () -> failure(either.left()),
            () -> success(either.right())))
    .forEach(Runnable::run);

Где failureи successкакие операции применять. Однако это создаст довольно много временных объектов и может быть не более эффективным, чем запуск из коллекции и потоковая / итерация дважды.

Другой способ обрабатывать элементы несколько раз - использовать Stream.peek (Consumer) :

doSomething().stream()
.peek(either -> handleFailure(either.left()))
.foreach(either -> handleSuccess(either.right()));

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

doSomething().stream()
.peek(element -> handleFoo(element.foo()))
.peek(element -> handleBar(element.bar()))
.peek(element -> handleBaz(element.baz()))
.foreach(element-> handleQux(element.qux()));

cyclops-react , библиотека, в которую я работаю, имеет статический метод, который позволит вам дублировать поток (и возвращает кортеж потоков jOOλ).

    Stream<Integer> stream = Stream.of(1,2,3);
    Tuple2<Stream<Integer>,Stream<Integer>> streams =  StreamUtils.duplicate(stream);

См. Комментарии, при использовании дубликата в существующем потоке будет снижена производительность. Более производительной альтернативой было бы использование Streamable: -

Существует также (ленивый) класс Streamable, который можно создать из Stream, Iterable или Array и многократно воспроизводить.

    Streamable<Integer> streamable = Streamable.of(1,2,3);
    streamable.stream().forEach(System.out::println);
    streamable.stream().forEach(System.out::println);

AsStreamable.synchronizedFromStream (stream) - может использоваться для создания Streamable, который будет лениво заполнять его резервную коллекцию таким образом, чтобы можно было совместно использовать его между потоками. Streamable.fromStream (stream) не повлечет за собой никаких накладных расходов на синхронизацию.

Для этой конкретной проблемы вы также можете использовать разделение. Что-то вроде

     // Partition Eighters into left and right
     List<Either<Pair<A, Throwable>, A>> results = doSomething();
     Map<Boolean, Object> passingFailing = results.collect(Collectors.partitioningBy(s -> s.isLeft()));
     passingFailing.get(true) <- here will be all passing (left values)
     passingFailing.get(false) <- here will be all failing (right values)

Мы можем использовать Stream Builder во время чтения или итерации потока. Вот документ Stream Builder .

https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.Builder.html

Пример использования

Допустим, у нас есть поток сотрудников, и нам нужно использовать этот поток для записи данных сотрудников в файл Excel, а затем обновить коллекцию / таблицу сотрудников [Это просто пример использования, демонстрирующий использование Stream Builder]:

Stream.Builder<Employee> builder = Stream.builder();

employee.forEach( emp -> {
   //store employee data to excel file 
   // and use the same object to build the stream.
   builder.add(emp);
});

//Now this stream can be used to update the employee collection
Stream<Employee> newStream = builder.build();

У меня была аналогичная проблема, и я мог подумать о трех различных промежуточных структурах, из которых можно было бы создать копию потока: a List, массив и a Stream.Builder. Я написал небольшую тестовую программу, которая показала, что с точки зрения производительности Listон примерно на 30% медленнее, чем два других, которые были довольно похожи.

Единственный недостаток преобразования в массив состоит в том, что это сложно, если ваш тип элемента является универсальным типом (что в моем случае было); поэтому я предпочитаю использовать Stream.Builder.

В итоге я написал небольшую функцию, которая создает Collector:

private static <T> Collector<T, Stream.Builder<T>, Stream<T>> copyCollector()
{
    return Collector.of(Stream::builder, Stream.Builder::add, (b1, b2) -> {
        b2.build().forEach(b1);
        return b1;
    }, Stream.Builder::build);
}

Затем я могу сделать копию любого потока str, сделав str.collect(copyCollector())это, как мне кажется, в соответствии с идиоматическим использованием потоков.