RxJava技术咨询：如何从源Observable中提取指定序列块？

可以用RxJava的操作符组合实现这个需求，核心思路是通过window操作符划分出以1开头、5结尾的窗口，再把窗口内的元素收集成列表。下面是具体实现：

实现步骤与代码示例 #

• 用window操作符，通过两个判断逻辑定义窗口的起始和结束条件：
  • 起始条件：元素等于1时开启新窗口
  • 结束条件：元素等于5时结束当前窗口
• 对每个窗口内的元素使用toList()收集成列表
• 过滤掉可能的空列表或不完整的块（比如序列末尾未闭合的1）

Observable<Integer> input = Observable.fromArray(1,2,3,2,4,5,0,2,1,3,4,5,2,2,1,4,5);

input.window(
        () -> input.filter(v -> v == 1), // 窗口开启信号：遇到1时开启新窗口
        v -> input.filter(s -> s == 5)   // 窗口关闭信号：遇到5时关闭当前窗口
    )
    .flatMap(window -> window.toList())
    .filter(list -> !list.isEmpty() && list.get(0) == 1 && list.get(list.size()-1) == 5)
    .subscribe(list -> System.out.println(list));

这段代码会输出：

[1, 2, 3, 2, 4, 5]
[1, 3, 4, 5]
[1, 4, 5]

操作符说明 #

• window：将原Observable拆分成多个窗口Observable，每个窗口对应一段符合起始/结束条件的元素序列
• flatMap + toList()：把每个窗口内的元素收集成单个列表对象，将窗口Observable转换为列表Observable
• filter：确保最终输出的列表都是完整的有效块（以1开头、5结尾），排除异常情况产生的无效列表

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