RxSwift技术需求：统计Observable中等价连续元素次数并压缩序列

这是个很实用的RxSwift场景，我之前处理过类似的需求。要把连续重复的元素转换成「元素+连续重复次数」的元组，我们可以通过组合几个基础操作符来实现，不需要第三方库，纯RxSwift原生API就能搞定。

实现方案 #

首先，给Observable写一个扩展方法，专门处理这个逻辑：

import RxSwift

extension Observable where Element: Equatable {
    func compressConsecutiveDuplicates() -> Observable<(Element, Int)> {
        return self
            // 用scan累积当前元素和连续计数
            .scan((current: Element?, count: 0)) { state, element in
                if state.current == element {
                    // 当前元素和累积元素相同，计数+1
                    return (element, state.count + 1)
                } else {
                    // 遇到新元素，重置计数为1
                    return (element, 1)
                }
            }
            .skip(1) // 跳过初始的空状态(nil, 0)
            // 在序列末尾添加哨兵元素，触发最后一组元素的发射
            .concat(Observable.just((nil, 0)))
            // 成对获取前后两个状态，判断元素是否变化
            .pairwise()
            // 元素变化时，发射前一组的统计结果
            .compactMap { previousState, nextState in
                if previousState.current != nextState.current {
                    return (previousState.current!, previousState.count)
                }
                return nil
            }
    }
}

代码解释 #

• scan操作符：用来维护一个「当前元素+连续计数」的状态，每次新元素到来时更新状态——和当前元素相同就累加计数，不同就重置为新元素和计数1。
• skip(1)：跳过scan一开始发射的初始空状态(nil, 0)，这个状态没有实际业务意义。
• concat(Observable.just((nil, 0)))：在原序列末尾添加一个哨兵状态，确保原序列结束时，最后一组连续元素的统计结果能被触发发射。
• pairwise()：把相邻的两个状态打包成元组，通过比较这两个状态的元素，判断是否进入了新的连续元素组。
• compactMap：只有当前后状态的元素不同时，才把前一个状态的「元素+计数」发射出去，这样就得到了每一组连续重复元素的统计结果。

测试示例 #

用你给出的输入序列来测试这个方法：

let disposeBag = DisposeBag()

// 输入序列
let source = Observable.of(1, 1, 1, 18, 21, 21, 0, -1, -8, -8, 14, 14, 14)

// 调用压缩方法并订阅
source.compressConsecutiveDuplicates()
    .subscribe(onNext: { element, count in
        print("(\(element), \(count))")
    })
    .disposed(by: disposeBag)

输出结果：

(1, 3)
(18, 1)
(21, 2)
(0, 1)
(-1, 1)
(-8, 2)
(14, 3)

完全符合你期望的转换结果。

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