如何将Java Stream或Reactor流中连续元素两两配对生成新对象

1. Java Stream 实现 #

Stream本身是无状态的单向迭代结构，原生没有直接提供相邻元素配对的能力，分两种场景实现：

1.1 有限有序流原生实现 #

如果你的流是有限、有序的，最简单的方式是先转成列表再按索引配对：

// 你的日期流
Stream<LocalDate> dateStream = Stream.of(LocalDate.of(2024,1,1), LocalDate.of(2024,1,2), LocalDate.of(2024,1,3));
// 定义配对对象，Java 16+可用record，低版本用普通类即可
record DatePair(LocalDate prev, LocalDate curr) {}

List<LocalDate> dateList = dateStream.toList();
Stream<DatePair> pairStream = IntStream.range(0, dateList.size() - 1)
        .mapToObj(i -> new DatePair(dateList.get(i), dateList.get(i + 1)));

注意：该方案仅适用于有限流，无限流调用toList()会永久阻塞。不要在map中用外部临时变量存储上一个元素的方式实现，并行流场景下会有线程安全问题。

1.2 通用实现（支持无限流/并行流） #

可以用增强Stream库StreamEx的内置pairMap方法实现，适配所有场景：

Stream<DatePair> pairStream = StreamEx.of(dateStream)
        .pairMap(DatePair::new);

2. Reactor Flux 实现 #

Reactor内置了相邻元素操作的能力，推荐两种写法：

2.1 推荐方案：zipWith + skip #

该写法性能更高，无多余计算：

Flux<LocalDate> dateFlux = Flux.just(LocalDate.of(2024,1,1), LocalDate.of(2024,1,2), LocalDate.of(2024,1,3));
record DatePair(LocalDate prev, LocalDate curr) {}

Flux<DatePair> pairFlux = dateFlux
        .zipWith(dateFlux.skip(1), DatePair::new);

2.2 缓存窗口实现 #

用buffer操作创建步长为1、大小为2的窗口，过滤掉最后不足2个元素的窗口即可：

Flux<DatePair> pairFlux = dateFlux
        .buffer(2, 1)
        .filter(list -> list.size() == 2)
        .map(list -> new DatePair(list.get(0), list.get(1)));

内容的提问来源于stack exchange，提问作者Shubhankar Agrawal