Java如何同时迭代/归约多个等长Stream流？

你不需要提前将所有Stream的全量元素收集到内存中，基于迭代器惰性拉取的特性就可以实现逐位运算，避免O(n)级别的存储开销。

核心思路 #

• 调用每个Stream的iterator()方法获取对应迭代器，迭代器不会预加载全量数据，只会在调用next()时拉取下一个元素
• 按位置索引从0到流长度逐轮遍历，每一轮从所有迭代器中各取出当前位置的1个元素，执行求和（或其他自定义运算）
• 全程仅持有当前轮次待计算的少量元素，内存开销为常数级，和单流总长度无关

测试验证用例 #

@Test
void aggregateMultipleStreams() {
    Stream<Integer> s1 = Stream.of(1, 2);
    Stream<Integer> s2 = Stream.of(4, 5);
    Stream<Integer> s3 = Stream.of(7, 8);
    assertEquals(List.of(1 + 4 + 7, 2 + 5 + 8), aggregate(s1, s2, s3, 2));
}

高开销的原有实现 #

原有实现先将每个Stream全量收集为List，当单流数据量极大时，会占用「流数量 * 单流长度」的内存空间，存储成本很高：

private List<Integer> aggregate(Stream<Integer> s1, Stream<Integer> s2, Stream<Integer> s3, int streamSize) {
    final List<List<Integer>> reduced = Stream.of(s1, s2, s3)
            .map(s -> s.collect(Collectors.toList())).collect(Collectors.toList());
    return IntStream.range(0, streamSize).mapToObj(n -> IntStream.range(0, reduced.size())
            .map(v -> reduced.get(v).get(n)).sum()).collect(Collectors.toList());
}

优化后无全量收集的实现 #

基于迭代器惰性拉取逻辑改造后，不需要预存全量元素即可完成按位求和：

private List<Integer> aggregate(Stream<Integer> s1, Stream<Integer> s2, Stream<Integer> s3, int streamSize) {
    final List<Iterator<Integer>> iterators = Stream.of(s1, s2, s3)
            .map(Stream::iterator).collect(Collectors.toList());
    return IntStream.range(0, streamSize).mapToObj(n -> IntStream.range(0, iterators.size())
            .map(v -> iterators.get(v).next()).sum()).collect(Collectors.toList());
}

实现前提：传入的所有Stream长度一致，且传入的streamSize参数和单流实际长度匹配，否则迭代器拉取时会抛出NoSuchElementException。如果需要支持任意数量流、自定义聚合规则，都可以基于这个惰性拉取的思路扩展。

内容的提问来源于stack exchange，提问作者Sendhilkumar Alalasundaram