Java 8 Stream，获取头尾

Java 8 Stream, getting head and tail

提问by lyomi

Java 8 introduced a Streamclass that resembles Scala's Stream, a powerful lazy construct using which it is possible to do something like this very concisely:

Java 8 引入了一个类似于 Scala 的Stream的Stream类，这是一个强大的惰性结构，使用它可以非常简洁地做这样的事情：

def from(n: Int): Stream[Int] = n #:: from(n+1)

def sieve(s: Stream[Int]): Stream[Int] = {
  s.head #:: sieve(s.tail filter (_ % s.head != 0))
}

val primes = sieve(from(2))

primes takeWhile(_ < 1000) print  // prints all primes less than 1000

I wondered if it is possible to do this in Java 8, so I wrote something like this:

我想知道是否可以在 Java 8 中做到这一点，所以我写了这样的东西：

IntStream from(int n) {
    return IntStream.iterate(n, m -> m + 1);
}

IntStream sieve(IntStream s) {
    int head = s.findFirst().getAsInt();
    return IntStream.concat(IntStream.of(head), sieve(s.skip(1).filter(n -> n % head != 0)));
}

IntStream primes = sieve(from(2));

Fairly simple, but it produces java.lang.IllegalStateException: stream has already been operated upon or closedbecause both findFirst()and skip()are terminal operations on Streamwhich can be done only once.

相当简单，但它会产生，java.lang.IllegalStateException: stream has already been operated upon or closed因为findFirst()和skip()都是终端操作，Stream只能执行一次。

I don't really have to use up the stream twice since all I need is the first number in the stream and the rest as another stream, i.e. equivalent of Scala's Stream.headand Stream.tail. Is there a method in Java 8 Streamthat I can use to achieve this?

我真的不必两次用完流，因为我只需要流中的第一个数字，其余的作为另一个流，即相当于 Scala 的Stream.head和Stream.tail。Stream我可以使用Java 8中的方法来实现这一点吗？

Thanks.

谢谢。

采纳答案by Holger

Even if you hadn't the problem that you can't split an IntStream, you code didn't work because you are invoking your sievemethod recursively instead of lazily. So you had an infinity recursion before you could query your resulting stream for the first value.

即使您没有无法拆分 的问题IntStream，您的代码也不起作用，因为您是sieve递归而不是懒惰地调用您的方法。因此，在查询结果流以获取第一个值之前，您进行了无限递归。

Splitting an IntStream sinto a head and a tail IntStream(which has not yet consumed) is possible:

将 an 拆分IntStream s为头部和尾部IntStream（尚未消耗）是可能的：

PrimitiveIterator.OfInt it = s.iterator();
int head = it.nextInt();
IntStream tail = IntStream.generate(it::next).filter(i -> i % head != 0);

At this place you need a construct of invoking sieveon the tail lazily. Streamdoes not provide that; concatexpects existing stream instances as arguments and you can't construct a stream invoking sievelazily with a lambda expression as lazy creation works with mutable state only which lambda expressions do not support. If you don't have a library implementation hiding the mutable state you have to use a mutable object. But once you accept the requirement of mutable state, the solution can be even easier than your first approach:

在这个地方，你需要一个sieve懒惰地调用尾部的构造。Stream不提供；concat期望现有的流实例作为参数，并且您不能构造一个sieve使用 lambda 表达式延迟调用的流，因为延迟创建仅适用于 lambda 表达式不支持的可变状态。如果您没有隐藏可变状态的库实现，则必须使用可变对象。但是一旦你接受了可变状态的要求，解决方案就会比你的第一种方法更容易：

IntStream primes = from(2).filter(i -> p.test(i)).peek(i -> p = p.and(v -> v % i != 0));

IntPredicate p = x -> true;

IntStream from(int n)
{
  return IntStream.iterate(n, m -> m + 1);
}

This will recursively create a filter but in the end it doesn't matter whether you create a tree of IntPredicates or a tree of IntStreams (like with your IntStream.concatapproach if it did work). If you don't like the mutable instance field for the filter you can hide it in an inner class (but not in a lambda expression…).

这将递归地创建一个过滤器，但最终创建一棵IntPredicates 树还是一棵IntStreams树并不重要（就像你的IntStream.concat方法一样，如果它确实有效）。如果您不喜欢过滤器的可变实例字段，您可以将其隐藏在内部类中（但不能隐藏在 lambda 表达式中……）。

回答by Lukas Eder

You can essentially implement it like this:

您基本上可以像这样实现它：

static <T> Tuple2<Optional<T>, Seq<T>> splitAtHead(Stream<T> stream) {
    Iterator<T> it = stream.iterator();
    return tuple(it.hasNext() ? Optional.of(it.next()) : Optional.empty(), seq(it));
}

In the above example, Tuple2and Seqare types borrowed from jOOλ, a library that we developed for jOOQintegration tests. If you don't want any additional dependencies, you might as well implement them yourself:

在上面的示例中，Tuple2和Seq是从jOOλ借用的类型，这是我们为jOOQ集成测试开发的库。如果你不想要任何额外的依赖，你也可以自己实现它们：

class Tuple2<T1, T2> {
    final T1 v1;
    final T2 v2;

    Tuple2(T1 v1, T2 v2) {
        this.v1 = v1;
        this.v2 = v2;
    }

    static <T1, T2> Tuple2<T1, T2> tuple(T1 v1, T2 v2) {
        return new Tuple<>(v1, v2);
    }
}

static <T> Tuple2<Optional<T>, Stream<T>> splitAtHead(Stream<T> stream) {
    Iterator<T> it = stream.iterator();
    return tuple(
        it.hasNext() ? Optional.of(it.next()) : Optional.empty,
        StreamSupport.stream(Spliterators.spliteratorUnknownSize(
            it, Spliterator.ORDERED
        ), false)
    );
}

回答by frhack

To get head and tail you need a Lazy Stream implementation. Java 8 stream or RxJava are not suitable.

要获得头部和尾部，您需要一个 Lazy Stream 实现。Java 8 流或 RxJava 不适合。

You can use for example LazySeqas follows.

例如，您可以使用如下LazySeq。

Lazy sequence is always traversed from the beginning using very cheap first/rest decomposition (head() and tail())

LazySeq implements java.util.List interface, thus can be used in variety of places. Moreover it also implements Java 8 enhancements to collections, namely streams and collectors

惰性序列总是从头开始遍历，使用非常便宜的 first/rest 分解（head() 和 tail()）

LazySeq 实现了 java.util.List 接口，因此可以在各种地方使用。此外，它还实现了 Java 8 对集合的增强，即流和收集器

---

---

package com.company;

import com.nurkiewicz.lazyseq.LazySeq;

public class Main {

    public static void main(String[] args) {

        LazySeq<Integer> ints = integers(2);
        LazySeq primes = sieve(ints);
        primes.take(10).forEach(p -> System.out.println(p));

    }

    private static LazySeq<Integer> sieve(LazySeq<Integer> s) {
        return LazySeq.cons(s.head(), () -> sieve(s.filter(x -> x % s.head() != 0)));
    }

    private static LazySeq<Integer> integers(int from) {
        return LazySeq.cons(from, () -> integers(from + 1));
    }

}

回答by frhack

Here is another recipe using the way suggested by Holger. It use RxJava just to add the possibility to use the take(int) method and many others.

这是另一个使用 Holger 建议的方法的食谱。它使用 RxJava 只是为了增加使用 take(int) 方法和许多其他方法的可能性。

package com.company;

import rx.Observable;

import java.util.function.IntPredicate;
import java.util.stream.IntStream;

public class Main {

    public static void main(String[] args) {

        final IntPredicate[] p={(x)->true};
        IntStream primesStream=IntStream.iterate(2,n->n+1).filter(i -> p[0].test(i)).peek(i->p[0]=p[0].and(v->v%i!=0)   );

        Observable primes = Observable.from(()->primesStream.iterator());

        primes.take(10).forEach((x) -> System.out.println(x.toString()));


    }

}

回答by vidi

The solution below does not do state mutations, except for the head/tail deconstruction of the stream.

下面的解决方案不做状态突变，除了流的头/尾解构。

The lazyness is obtained using IntStream.iterate. The class Prime is used to keep the generator state

惰性是使用 IntStream.iterate 获得的。类 Prime 用于保持生成器状态

    import java.util.PrimitiveIterator;
    import java.util.stream.IntStream;
    import java.util.stream.Stream;

    public class Prime {
        private final IntStream candidates;
        private final int current;

        private Prime(int current, IntStream candidates)
        {
            this.current = current;
            this.candidates = candidates;
        }

        private Prime next()
        {
            PrimitiveIterator.OfInt it = candidates.filter(n -> n % current != 0).iterator();

            int head = it.next();
            IntStream tail = IntStream.generate(it::next);

            return new Prime(head, tail);
        }

        public static Stream<Integer> stream() {
            IntStream possiblePrimes = IntStream.iterate(3, i -> i + 1);

            return Stream.iterate(new Prime(2, possiblePrimes), Prime::next)
                         .map(p -> p.current);
        }
    }

The usage would be this:

用法是这样的：

Stream<Integer> first10Primes = Prime.stream().limit(10)

回答by John McClean

If you don't mind using 3rd party libraries cyclops-streams, I library I wrote has a number of potential solutions.

如果您不介意使用第 3 方库cyclops-streams，我编写的库有许多潜在的解决方案。

The StreamUtilsclass has large number of static methods for working directly with java.util.stream.Streamsincluding headAndTail.

该StreamUtils类有大量的静态方法直接与工作java.util.stream.Streams包括headAndTail。

HeadAndTail<Integer> headAndTail = StreamUtils.headAndTail(Stream.of(1,2,3,4));
int head = headAndTail.head(); //1
Stream<Integer> tail = headAndTail.tail(); //Stream[2,3,4]

The Streamableclass represents a replayable Streamand works by building a lazy, caching intermediate data-structure. Because it is caching and repayable - head and tail can be implemented directly and separately.

该可流化类表示一重玩Stream并通过建立一个懒惰，缓存中间数据结构的工作原理。因为它是缓存和可偿还的——head和tail可以直接分开实现。

Streamable<Integer> replayable=  Streamable.fromStream(Stream.of(1,2,3,4));
int head = repayable.head(); //1
Stream<Integer> tail = replayable.tail(); //Stream[2,3,4]

cyclops-streamsalso provides a sequential Streamextension that in turn extends jOOλand has both Tuplebased (from jOOλ) and domain object (HeadAndTail) solutions for head and tail extraction.

cyclops-streams还提供了一个顺序Stream扩展，依次扩展了jOOλ，并且具有Tuple用于头部和尾部提取的基于（来自 jOOλ）和域对象（HeadAndTail）的解决方案。

SequenceM.of(1,2,3,4)
         .splitAtHead(); //Tuple[1,SequenceM[2,3,4]

SequenceM.of(1,2,3,4)
         .headAndTail();

Update per Tagir's request -> A Java version of the Scala sieve using SequenceM

根据 Tagir 的请求更新 -> Scala 筛的 Java 版本使用 SequenceM

public void sieveTest(){
    sieve(SequenceM.range(2, 1_000)).forEach(System.out::println);
}

SequenceM<Integer> sieve(SequenceM<Integer> s){

    return s.headAndTailOptional().map(ht ->SequenceM.of(ht.head())
                            .appendStream(sieve(ht.tail().filter(n -> n % ht.head() != 0))))
                    .orElse(SequenceM.of());
}

And another version via Streamable

另一个版本通过 Streamable

public void sieveTest2(){
    sieve(Streamable.range(2, 1_000)).forEach(System.out::println);
}

Streamable<Integer> sieve(Streamable<Integer> s){

    return s.size()==0? Streamable.of() : Streamable.of(s.head())
                                                    .appendStreamable(sieve(s.tail()
                                                                    .filter(n -> n % s.head() != 0)));
}

Note - neither Streamableof SequenceMhave an Empty implementation - hence the size check for Streamableand the use of headAndTailOptional.

注-既不Streamable的SequenceM有一个空的执行-因此尺寸检查Streamable和使用headAndTailOptional。

Finally a version using plain java.util.stream.Stream

最后一个使用普通的版本 java.util.stream.Stream

import static com.aol.cyclops.streams.StreamUtils.headAndTailOptional;

public void sieveTest(){
    sieve(IntStream.range(2, 1_000).boxed()).forEach(System.out::println);
}

Stream<Integer> sieve(Stream<Integer> s){

    return headAndTailOptional(s).map(ht ->Stream.concat(Stream.of(ht.head())
                            ,sieve(ht.tail().filter(n -> n % ht.head() != 0))))
                    .orElse(Stream.of());
}

Another update - a lazy iterative based on @Holger's version using objects rather than primitives (note a primitive version is also possible)

另一个更新 - 基于@Holger 版本的惰性迭代，使用对象而不是基元（注意基元版本也是可能的）

  final Mutable<Predicate<Integer>> predicate = Mutable.of(x->true);
  SequenceM.iterate(2, n->n+1)
           .filter(i->predicate.get().test(i))
           .peek(i->predicate.mutate(p-> p.and(v -> v%i!=0)))
           .limit(100000)
           .forEach(System.out::println);

回答by Tagir Valeev

My StreamExlibrary has now headTail()operation which solves the problem:

我的StreamEx库现在headTail()有解决问题的操作：

public static StreamEx<Integer> sieve(StreamEx<Integer> input) {
    return input.headTail((head, tail) -> 
        sieve(tail.filter(n -> n % head != 0)).prepend(head));
}

The headTailmethod takes a BiFunctionwhich will be executed at most once during the stream terminal operation execution. So this implementation is lazy: it does not compute anything until traversal starts and computes only as much prime numbers as requested. The BiFunctionreceives a first stream element headand the stream of the rest elements tailand can modify the tailin any way it wants. You may use it with predefined input:

该headTail方法需要BiFunction在流终端操作执行期间最多执行一次。所以这个实现是惰性的：它在遍历开始之前不计算任何东西，并且只计算所要求的素数。所述BiFunction接收第一数据流元件head和其余元件的流tail，并且可以修改tail在它想要的任何方式。您可以将它与预定义的输入一起使用：

sieve(IntStreamEx.range(2, 1000).boxed()).forEach(System.out::println);

But infinite stream work as well

但无限流也能工作

sieve(StreamEx.iterate(2, x -> x+1)).takeWhile(x -> x < 1000)
     .forEach(System.out::println);
// Not the primes till 1000, but 1000 first primes
sieve(StreamEx.iterate(2, x -> x+1)).limit(1000).forEach(System.out::println);

There's also alternative solution using headTailand predicate concatenation:

还有使用headTail和谓词连接的替代解决方案：

public static StreamEx<Integer> sieve(StreamEx<Integer> input, IntPredicate isPrime) {
    return input.headTail((head, tail) -> isPrime.test(head) 
            ? sieve(tail, isPrime.and(n -> n % head != 0)).prepend(head)
            : sieve(tail, isPrime));
}

sieve(StreamEx.iterate(2, x -> x+1), i -> true).limit(1000).forEach(System.out::println);

---

---

It interesting to compare recursive solutions: how many primes they capable to generate.

比较递归解决方案很有趣：它们能够生成多少素数。

@John McClean solution (StreamUtils)

@John McClean 解决方案 ( StreamUtils)

John McClean solutions are not lazy: you cannot feed them with infinite stream. So I just found by trial-and-error the maximal allowed upper bound (17793) (after that StackOverflowError occurs):

John McClean 的解决方案并不懒惰：你不能用无限流喂它们。所以我只是通过反复试验发现最大允许上限 ( 17793) （在发生 StackOverflowError 之后）：

public void sieveTest(){
    sieve(IntStream.range(2, 17793).boxed()).forEach(System.out::println);
}

@John McClean solution (Streamable)

@John McClean 解决方案 ( Streamable)

public void sieveTest2(){
    sieve(Streamable.range(2, 39990)).forEach(System.out::println);
}

Increasing upper limit above 39990results in StackOverflowError.

增加上限会39990导致 StackOverflowError。

@frhack solution (LazySeq)

@frhack 解决方案 ( LazySeq)

LazySeq<Integer> ints = integers(2);
LazySeq primes = sieve(ints); // sieve method from @frhack answer
primes.forEach(p -> System.out.println(p));

Result: stuck after prime number = 53327with enormous heap allocation and garbage collection taking more than 90%. It took several minutes to advance from 53323 to 53327, so waiting more seems impractical.

结果：在质数之后卡住 =53327巨大的堆分配和垃圾收集占用超过 90%。从 53323 到 53327 需要几分钟的时间，所以等待更多似乎不切实际。

@vidi solution

@vidi 解决方案

Prime.stream().forEach(System.out::println);

Result: StackOverflowError after prime number = 134417.

结果：质数 = 后出现 StackOverflowError 134417。

My solution (StreamEx)

我的解决方案（StreamEx）

sieve(StreamEx.iterate(2, x -> x+1)).forEach(System.out::println);

Result: StackOverflowError after prime number = 236167.

结果：质数 = 后出现 StackOverflowError 236167。

@frhack solution (rxjava)

@frhack 解决方案 ( rxjava)

Observable<Integer> primes = Observable.from(()->primesStream.iterator());
primes.forEach((x) -> System.out.println(x.toString()));            

Result: StackOverflowError after prime number = 367663.

结果：质数 = 后出现 StackOverflowError 367663。

@Holger solution

@霍尔格解决方案

IntStream primes=from(2).filter(i->p.test(i)).peek(i->p=p.and(v->v%i!=0));
primes.forEach(System.out::println);

Result: StackOverflowError after prime number = 368089.

结果：质数 = 后出现 StackOverflowError 368089。

My solution (StreamEx with predicate concatenation)

我的解决方案（带有谓词串联的 StreamEx）

sieve(StreamEx.iterate(2, x -> x+1), i -> true).forEach(System.out::println);

Result: StackOverflowError after prime number = 368287.

结果：质数 = 后出现 StackOverflowError 368287。

---

---

So three solutions involving predicate concatenation win, because each new condition adds only 2 more stack frames. I think, the difference between them is marginal and should not be considered to define a winner. However I like my first StreamEx solution more as it more similar to Scala code.

所以涉及谓词串联的三个解决方案获胜，因为每个新条件只增加了 2 个堆栈帧。我认为，它们之间的差异很小，不应被视为定义赢家。但是我更喜欢我的第一个 StreamEx 解决方案，因为它更类似于 Scala 代码。

回答by user_3380739

If you want to get head of a stream, just:

如果你想获得一个流的头部，只需：

IntStream.range(1, 5).first();

If you want to get tail of a stream, just:

如果您想获得流的尾部，只需：

IntStream.range(1, 5).skip(1);

If you want to get both head and tail of a stream, just:

如果您想同时获得流的头部和尾部，只需：

IntStream s = IntStream.range(1, 5);
int head = s.head();
IntStream tail = s.tail();

If you want to find the prime, just:

如果你想找到素数，只需：

LongStream.range(2, n)
   .filter(i -> LongStream.range(2, (long) Math.sqrt(i) + 1).noneMatch(j -> i % j == 0))
   .forEach(N::println);

If you want to know more, go to get AbacusUtil

如果你想了解更多，去获取AbacusUtil

Declaration： I'm the developer of AbacusUtil.

声明： 我是 AbacusUtil 的开发者。

回答by Peter

There are many interesting suggestions provided here, but if someone needs a solution without dependencies to third party libraries I came up with this:

这里提供了许多有趣的建议，但是如果有人需要一个不依赖第三方库的解决方案，我想出了这个：

    import java.util.AbstractMap;
    import java.util.Optional;
    import java.util.Spliterators;
    import java.util.stream.StreamSupport;

    /**
     * Splits a stream in the head element and a tail stream.
     * Parallel streams are not supported.
     * 
     * @param stream Stream to split.
     * @param <T> Type of the input stream.
     * @return A map entry where {@link Map.Entry#getKey()} contains an
     *    optional with the first element (head) of the original stream
     *    and {@link Map.Entry#getValue()} the tail of the original stream.
     * @throws IllegalArgumentException for parallel streams.
     */
    public static <T> Map.Entry<Optional<T>, Stream<T>> headAndTail(final Stream<T> stream) {
        if (stream.isParallel()) {
            throw new IllegalArgumentException("parallel streams are not supported");
        }
        final Iterator<T> iterator = stream.iterator();
        return new AbstractMap.SimpleImmutableEntry<>(
                iterator.hasNext() ? Optional.of(iterator.next()) : Optional.empty(),
                StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, 0), false)
        );
    }