могу ли я продублировать поток в java 8?

<p>иногда я хочу выполнить набор операций над потоком, а затем обработать полученный поток двумя различными способами с помощью других операций.</p><p>могу ли я сделать это без указания общих начальных операций дважды?</p><p>например, я надеюсь, что существует метод <code>dup()</code>, такой как:</p><pre><code>stream [] desired\_streams = intstream.range(1, 100).filter(n -> n % 2 == 0).dup(); stream stream14 = desired\_streams[0].filter(n -> n % 7 == 0); // multiples of 14 stream stream10 = desired\_streams[1].filter(n -> n % 5 == 0); // multiples of 10 </code></pre>

<p>это невозможно вообще.</p><p>если вы хотите дублировать поток ввода или итератор ввода, у вас есть два варианта:</p><h3>a. храните все в коллекции, скажем, <code>list<></code></h3><p>предположим, вы дублируете поток в два потока <code>s1</code> и <code>s2</code>. если у вас есть расширенные элементы <code>n1</code> в элементах <code>s1</code> и <code>n2</code> с <code>s2</code>, вы должны хранить элементы <code>|n2 - n1|</code> в памяти, просто чтобы идти в ногу. если ваш поток бесконечный, может не быть верхней границы для требуемого хранилища.</p><p>взгляните на python <a href="https://docs.python.org/3/library/itertools.html#itertools.tee"><code>tee()</code></a> , чтобы увидеть, что нужно:</p><blockquote> <p>этот itertool может потребовать значительного вспомогательного хранения (в зависимости от того, сколько временных данных необходимо сохранить). в общем, если один итератор использует большую часть или все данные до запуска другого итератора, быстрее использовать <code>list()</code> вместо <code>tee()</code>.</p> </blockquote><h3>b. когда возможно: скопируйте состояние генератора, который создает элементы</h3><p>чтобы эта опция работала, вам, вероятно, понадобится доступ к внутренней работе потока. другими словами, генератор - часть, которая создает элементы - должен в первую очередь поддерживать копирование. [op: смотрите это <a href="https://stackoverflow.com/a/24474871/621338"><strong>отличный ответ</strong></a> , как пример того, как это можно сделать для примера в вопросе]</p><p>он не будет работать при вводе данных пользователем, поскольку вам придется копировать состояние всего «внешнего мира». java <code>stream</code> не поддерживает копирование, поскольку оно разработано так, чтобы быть максимально общим, особенно для работы с файлами, сетью, клавиатурой, датчиками, случайностью и т.д. [op: другой пример - поток, который читает датчик температуры по требованию. это не может быть продублировано без сохранения копии показаний]</p><p>это не только случай в java; это общее правило. вы можете видеть, что <a href="http://www.cplusplus.com/reference/istream/istream/istream/"><code>std::istream</code></a> в c++ поддерживает только семантику перемещения, но не семантику копирования («конструктор копирования (удаленный)»), по этой причине (и другим). </p>

<p>таким способом невозможно дублировать поток. однако вы можете избежать дублирования кода, переместив общую часть в метод или лямбда-выражение.</p><pre><code>supplier<intstream> supplier = () -> intstream.range(1, 100).filter(n -> n % 2 == 0); supplier.get().filter(...); supplier.get().filter(...); </code></pre>

<p>это возможно, если вы буферизуете элементы, которые вы использовали в одном дубликате, но еще не в другом.</p><p>мы реализовали метод <code>duplicate()</code> для потоков в <a href="https://github.com/jooq/jool">jooλ</a> , библиотеке с открытым исходным кодом, которую мы создали, чтобы улучшить интеграционное тестирование для <a href="http://www.jooq.org">jooq</a> . по сути, вы можете просто написать:</p><pre><code>tuple2<seq<integer>, seq<integer>> desired\_streams = seq.seq( intstream.range(1, 100).filter(n -> n % 2 == 0).boxed() ).duplicate(); </code></pre><p><sub>(примечание: в настоящее время нам нужно заблокировать поток, так как мы еще не реализовали переменную <code>intseq</code>)</sub></p><p>внутри есть буфер <code>linkedlist</code>, в котором хранятся все значения, которые были получены из одного потока, но не из другого. это, вероятно, так же эффективно, как если бы два ваших потока потреблялись примерно с одинаковой скоростью.</p><p>вот как работает алгоритм:</p><pre><code>static <t> tuple2<seq<t>, seq<t>> duplicate(stream<t> stream) { final linkedlist<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())); } </code></pre><p> <a href="https://github.com/jooq/jool/blob/master/src/main/java/org/jooq/lambda/seq.java">больше исходного кода здесь</a> </p><p>фактически, используя <a href="https://github.com/jooq/jool">jooλ</a> , вы сможете написать полный однострочный текст примерно так:</p><pre><code>tuple2<seq<integer>, seq<integer>> desired\_streams = seq.seq( intstream.range(1, 100).filter(n -> n % 2 == 0).boxed() ).duplicate() .map1(s -> s.filter(n -> n % 7 == 0)) .map2(s -> s.filter(n -> n % 5 == 0)); // this will yield 14, 28, 42, 56... desired\_streams.v1.foreach(system.out::println) // this will yield 10, 20, 30, 40... desired\_streams.v2.foreach(system.out::println); </code></pre>

<p>вы также можете переместить генерацию потока в отдельный метод/функцию, которая возвращает этот поток, и вызвать его дважды.</p>

<p>или,</p><ul> <li>переместите инициализацию в метод и просто вызовите метод снова</li> </ul><p>преимущество этого в том, что вы явно говорите о том, что вы делаете, а также работает для бесконечных потоков.</p><ul> <li>соберите поток и затем повторно направьте его</li> </ul><p>в вашем примере:</p><pre><code>final int[] arr = intstream.range(1, 100).filter(n -> n % 2 == 0).toarray(); </code></pre><p>затем</p><pre><code>final intstream s = intstream.of(arr); </code></pre>

<p><strong> обновление: </strong>это<strong> не </strong>/ работает. см. объяснение ниже, после текста оригинального ответа.</p><p>как глупо с моей стороны. все, что мне нужно сделать, это:</p><pre><code>stream desired\_stream = intstream.range(1, 100).filter(n -> n % 2 == 0); stream stream14 = desired\_stream.filter(n -> n % 7 == 0); // multiples of 14 stream stream10 = desired\_stream.filter(n -> n % 5 == 0); // multiples of 10 </code></pre><p><strong> объяснение, почему это не работает: </strong></p><p>если вы закодируете его и попытаетесь собрать оба потока, первый будет собираться нормально, но при попытке выполнить потоковую передачу будет выдано исключение: <code>java.lang.illegalstateexception: stream has already been operated upon or closed</code>.</p><p>чтобы уточнить, потоки являются объектами с состоянием (которые, кстати, не могут быть сброшены или перемотаны). вы можете думать о них как об итераторах, которые, в свою очередь, похожи на указатели. таким образом, <code>stream14</code> и <code>stream10</code> можно рассматривать как ссылки на один и тот же указатель. полное использование первого потока приведет к тому, что указатель пойдет «мимо конца». попытка использовать второй поток - это все равно, что пытаться получить доступ к указателю, который уже «за пределами конца», что, естественно, является недопустимой операцией.</p><p>как показывает принятый ответ, код для создания потока должен выполняться дважды, но его можно разделить на лямбду <code>supplier</code> или аналогичную конструкцию.</p><p><strong> полный тестовый код: </strong>сохранить в <code>foo.java</code>, затем <code>javac foo.java</code>, затем <code>java foo</code></p><pre><code>import java.util.stream.intstream; public class foo { public static void main (string [] args) { intstream s = intstream.range(0, 100).filter(n -> n % 2 == 0); intstream s1 = s.filter(n -> n % 5 == 0); s1.foreach(n -> system.out.println(n)); intstream s2 = s.filter(n -> n % 7 == 0); s2.foreach(n -> system.out.println(n)); } } </code></pre><p><strong> результат: </strong></p><pre><code>$ javac foo.java $ java foo 0 10 20 30 40 50 60 70 80 90 exception in thread "main" java.lang.illegalstateexception: stream has already been operated upon or closed at java.util.stream.abstractpipeline.<init>(abstractpipeline.java:203) at java.util.stream.intpipeline.<init>(intpipeline.java:91) at java.util.stream.intpipeline$statelessop.(intpipeline.java:592) at java.util.stream.intpipeline$9.<init>(intpipeline.java:332) at java.util.stream.intpipeline.filter(intpipeline.java:331) at foo.main(foo.java:8) </code></pre>

<p>для бесконечных потоков, если у вас есть доступ к источнику, это прямо:</p><pre><code>@test public void testname() throws exception { list<integer> integers = arrays.aslist(1, 2, 4, 5, 6, 7, 8, 9, 10); stream<integer> stream1 = integers.stream(); stream<integer> stream2 = integers.stream(); stream1.foreach(system.out::println); stream2.foreach(system.out::println); } </code></pre><blockquote> <p>печать</p> <p>1 2 4 5 6 7 8 9 10</p> <p>1 2 4 5 6 7 8 9 10</p> </blockquote><p>для вашего случая:</p><pre><code>stream originalstream = intstream.range(1, 100).filter(n -> n % 2 == 0) list<integer> listof = originalstream.collect(collectors.tolist()) stream stream14 = listof.stream().filter(n -> n % 7 == 0); stream stream10 = listof.stream().filter(n -> n % 5 == 0); </code></pre><p>для производительности и т.д. читайте чужой ответ;)</p>