从流中收集连续的对

Question

鉴于如下的流{ 0, 1, 2, 3, 4 },

我怎样才能最优雅地将其转换为给定的形式:

{ new Pair(0, 1), new Pair(1, 2), new Pair(2, 3), new Pair(3, 4) }

(假设,当然,我已经定义了类对)？

编辑:这不是严格关于整数或原始流.对于任何类型的流,答案应该是通用的.

Answer 1

Java 8流库主要用于将流拆分为较小的块以进行并行处理,因此有状态的流水线阶段非常有限,并且不支持获取当前流元素的索引和访问相邻流元素等操作.

解决这些问题的典型方法当然有一些限制,即通过索引驱动流并依赖于在某些随机访问数据结构中处理值,例如可以从中检索元素的ArrayList.如果值为arrayList,则可以通过执行以下操作生成所请求的对:

    IntStream.range(1, arrayList.size())
             .mapToObj(i -> new Pair(arrayList.get(i-1), arrayList.get(i)))
             .forEach(System.out::println);

当然,限制是输入不能是无限流.但是,此管道可以并行运行.

Answer 2

我的StreamEx库扩展了标准流,pairMap为所有流类型提供了一种方法.对于原始流,它不会更改流类型,但可用于进行一些计算.最常见的用法是计算差异:

int[] pairwiseDiffs = IntStreamEx.of(input).pairMap((a, b) -> (b-a)).toArray();

对于对象流,您可以创建任何其他对象类型.我的库没有提供任何新的用户可见数据结构Pair(这是库概念的一部分).但是,如果您有自己的Pair类并想要使用它,则可以执行以下操作:

Stream<Pair> pairs = IntStreamEx.of(input).boxed().pairMap(Pair::new);

或者,如果你已经有一些Stream:

Stream<Pair> pairs = StreamEx.of(stream).pairMap(Pair::new);

此功能使用自定义spliterator实现.它的开销很低,可以很好地并行化.当然它适用于任何流源,而不仅仅是随机访问列表/数组,就像许多其他解决方案一样.在许多测试中,它表现得非常好.这是一个JMH基准测试,我们使用不同的方法找到更大值之前的所有输入值(参见此问题).

Answer 3

这不是优雅,它是一个hackish解决方案,但适用于无限流

Stream<Pair> pairStream = Stream.iterate(0, (i) -> i + 1).map( // natural numbers
    new Function<Integer, Pair>() {
        Integer previous;

        @Override
        public Pair apply(Integer integer) {
            Pair pair = null;
            if (previous != null) pair = new Pair(previous, integer);
            previous = integer;
            return pair;
        }
    }).skip(1); // drop first null

现在,您可以将流限制为所需的长度

pairStream.limit(1_000_000).forEach(i -> System.out.println(i));

PS我希望有更好的解决方案,比如clojure(partition 2 1 stream)

Answer 4

我已经实现了一个spliterator包装器,它从原始的spliterator中获取每个n元素T并生成List<T>:

public class ConsecutiveSpliterator<T> implements Spliterator<List<T>> {

    private final Spliterator<T> wrappedSpliterator;

    private final int n;

    private final Deque<T> deque;

    private final Consumer<T> dequeConsumer;

    public ConsecutiveSpliterator(Spliterator<T> wrappedSpliterator, int n) {
        this.wrappedSpliterator = wrappedSpliterator;
        this.n = n;
        this.deque = new ArrayDeque<>();
        this.dequeConsumer = deque::addLast;
    }

    @Override
    public boolean tryAdvance(Consumer<? super List<T>> action) {
        deque.pollFirst();
        fillDeque();
        if (deque.size() == n) {
            List<T> list = new ArrayList<>(deque);
            action.accept(list);
            return true;
        } else {
            return false;
        }
    }

    private void fillDeque() {
        while (deque.size() < n && wrappedSpliterator.tryAdvance(dequeConsumer))
            ;
    }

    @Override
    public Spliterator<List<T>> trySplit() {
        return null;
    }

    @Override
    public long estimateSize() {
        return wrappedSpliterator.estimateSize();
    }

    @Override
    public int characteristics() {
        return wrappedSpliterator.characteristics();
    }
}

以下方法可用于创建连续流:

public <E> Stream<List<E>> consecutiveStream(Stream<E> stream, int n) {
    Spliterator<E> spliterator = stream.spliterator();
    Spliterator<List<E>> wrapper = new ConsecutiveSpliterator<>(spliterator, n);
    return StreamSupport.stream(wrapper, false);
}

样品用法:

consecutiveStream(Stream.of(0, 1, 2, 3, 4, 5), 2)
    .map(list -> new Pair(list.get(0), list.get(1)))
    .forEach(System.out::println);

Answer 5

您可以使用Stream.reduce()方法执行此操作(我没有看到使用此技术的任何其他答案).

public static <T> List<Pair<T, T>> consecutive(List<T> list) {
    List<Pair<T, T>> pairs = new LinkedList<>();
    list.stream().reduce((a, b) -> {
        pairs.add(new Pair<>(a, b));
        return b;
    });
    return pairs;
}

Answer 6

您可以使用滑动操作符在cyclops-react(我参与此库)中执行此操作.

  LazyFutureStream.of( 0, 1, 2, 3, 4 )
                  .sliding(2)
                  .map(Pair::new);

要么

   ReactiveSeq.of( 0, 1, 2, 3, 4 )
                  .sliding(2)
                  .map(Pair::new);

假设Pair构造函数可以接受包含2个元素的Collection.

如果要按4分组,则增加2,也支持.

     ReactiveSeq.rangeLong( 0L,Long.MAX_VALUE)
                .sliding(4,2)
                .forEach(System.out::println);

在cyclops-streams StreamUtils类中还提供了用于在java.util.stream.Stream上创建滑动视图的Equivalant静态方法.

       StreamUtils.sliding(Stream.of(1,2,3,4),2)
                  .map(Pair::new);

注意: - 对于单线程操作,ReactiveSeq更合适.LazyFutureStream扩展了ReactiveSeq,但主要用于并发/并行使用(它是一个期货流).

LazyFutureStream扩展了ReactiveSeq,它从令人敬畏的jOOλ(扩展了java.util.stream.Stream)扩展了Seq,因此Lukas提供的解决方案也适用于Stream类型.对于任何感兴趣的人来说,窗口/滑动运算符之间的主要区别在于明显的相对功率/复杂度权衡以及与无限流一起使用的适合性(滑动不消耗流,而是流动时的缓冲区).

Answer 7

proton-pack 库提供窗口功能。给定一个 Pair 类和一个 Stream，你可以这样做：

Stream<Integer> st = Stream.iterate(0 , x -> x + 1);
Stream<Pair<Integer, Integer>> pairs = StreamUtils.windowed(st, 2, 1)
                                                  .map(l -> new Pair<>(l.get(0), l.get(1)))
                                                  .moreStreamOps(...);

现在该pairs流包含：

(0, 1)
(1, 2)
(2, 3)
(3, 4)
(4, ...) and so on

Answer 8

寻找连续的对

\n\n

如果您愿意使用第三方库并且不需要并行性，那么jOO\xce\xbb提供 SQL 风格的窗口函数，如下所示

\n\n

System.out.println(\nSeq.of(0, 1, 2, 3, 4)\n   .window()\n   .filter(w -> w.lead().isPresent())\n   .map(w -> tuple(w.value(), w.lead().get())) // alternatively, use your new Pair() class\n   .toList()\n);\n

\n\n

屈服

\n\n\n\n

[(0, 1), (1, 2), (2, 3), (3, 4)]\n

\n\n

这lead()函数按遍历顺序从窗口访问下一个值。

\n\n

查找连续的三元组/四元组/n元组

\n\n

评论中的一个问题是要求一个更通用的解决方案，其中不应收集对，而是应收集 n 元组（或可能是列表）。因此，这是一种替代方法：

\n\n

[(0, 1), (1, 2), (2, 3), (3, 4)]\n

\n\n

生成列表列表

\n\n\n\n

[[0, 1, 2], [1, 2, 3], [2, 3, 4]]\n

\n\n

如果没有filter(w -> w.count() == n)，结果将是

\n\n\n\n

[[0, 1, 2], [1, 2, 3], [2, 3, 4], [3, 4], [4]]\n

\n\n

免责声明：我在 jOO\xce\xbb 背后的公司工作

\n

Answer 9

Streams.zip(..)在 Guava 中可用，对于那些依赖它的人。

例子：

Streams.zip(list.stream(),
            list.stream().skip(1),
            (a, b) -> System.out.printf("%s %s\n", a, b));

Answer 10

JEP 461：Stream Gatherers Java 22 预览语言功能对此有内置支持：

\n

Stream.of(0, 1, 2, 3, 4)\n        .gather(Gatherers.windowSliding(2))\n        .map(list -> new Pair(list.get(0), list.get(1)))\n        .toList();\n

\n

这使用新Stream.gather方法和新的内置Gatherers.windowSliding收集器将初始Stream<Integer>( [0, 1, 2, 3, 4]) 转换为成对Stream<List<Integer>>( [[0, 1], [1, 2], [2, 3], [3, 4]])。Pair然后使用现有方法将这些列表中的每一个转换为列表Stream.map。

\n

Java文档

\n

Gatherer:

\n

\n

将输入元素流转换为输出元素流的中间操作，可以选择在到达上游末尾时应用最终操作。[\xe2\x80\xa6]

\n

[\xe2\x80\xa6]

\n

收集操作的示例有很多，包括但不限于：将元素分组为批次（窗口函数）；对连续相似的元素进行去重；增量累加功能（前缀扫描）；增量重新排序功能等。该类Gatherers提供常见收集操作的实现。

\n

\n

Stream.gather:

\n

\n

返回一个流，其中包含将给定收集器应用于该流的元素的结果。

\n

\n

Gatherers.windowSliding

\n

\n

返回一个 Gatherer，它将元素收集到给定大小的窗口（按遇到顺序排列的元素组）中，其中每个后续窗口都包含前一个窗口的所有元素（除了最近的元素之外），并在流中添加下一个元素。[\xe2\x80\xa6]

\n

例子：

\n

// will contain: [[1, 2], [2, 3], [3, 4], [4, 5], [5, 6], [6, 7], [7, 8]]\nList<List<Integer>> windows2 =\n    Stream.of(1,2,3,4,5,6,7,8).gather(Gatherers.windowSliding(2)).toList();\n\n// will contain: [[1, 2, 3, 4, 5, 6], [2, 3, 4, 5, 6, 7], [3, 4, 5, 6, 7, 8]]\nList<List<Integer>> windows6 =\n    Stream.of(1,2,3,4,5,6,7,8).gather(Gatherers.windowSliding(6)).toList();\n

\n

\n