RxJava2/RxKotlin2：如何将Observable所有发射值聚合为单个结果对象

原实现的核心问题 #

最初基于scan的实现存在两个和需求不匹配的核心问题：

• 操作符特性错配：scan 的行为是每接收到一个上游发射的元素，就立刻向下游发射一次当前的中间聚合值，会持续输出多轮结果，并不会等待全量元素处理完只输出一个最终聚合结果
• 代码与状态设计风险：原代码的lambda参数标注位置不符合Kotlin语法规则，无法正常编译；同时原逻辑中直接修改传入的Result实例的可变字段，既违反RxJava推荐的不可变状态约定，也容易因为操作符内部的引用共享，导致多次订阅时出现状态污染

适配需求的最优实现：使用reduce操作符 #

如果核心目标是等Observable把所有Trade元素全部发射完成后，仅输出一份全量聚合的最终Result，直接使用reduce操作符是最匹配的方案。
reduce的行为和需求完全对齐：内部会持续累加处理每一个上游元素，直到上游调用onComplete通知流结束时，才把最终的累计结果作为唯一值发射给下游。
由于本次场景中上游发射的元素类型（Trade）和最终累计值类型（Result）不一致，需要调用带初始值参数的reduce重载，传入空的Result作为聚合起点。

参考代码实现 #

首先推荐将数据类定义为不可变结构，从根源上避免可变状态导致的bug：

data class Trade(
    val tradeType: String,
    val profit: BigDecimal
)

data class Result(
    val totalProfit: BigDecimal = BigDecimal.ZERO
)

基于reduce的聚合逻辑：

trades
    // 传入空的Result作为聚合的初始累加值
    .reduce(
        initialValue = Result(),
        accumulator = { currentResult: Result, newTrade: Trade ->
            // 每次生成全新的Result实例存储累计值，不修改原有对象
            Result(
                totalProfit = currentResult.totalProfit + newTrade.profit
            )
        }
    )
    .subscribe(
        { finalResult -> println("最终累计总收益：${finalResult.totalProfit}") },
        { throwable -> throwable.printStackTrace() }
    )

操作符选型参考 #

• 如果需要实时展示累计进度（比如边接收交易流边更新界面上的总收益数字），可以使用scan操作符，它会持续发射每一步的中间累计结果，满足实时更新的需求
• 如果只需要全量处理完成后的最终聚合值，直接使用reduce即可，没有不必要的中间值发射开销；如果非要基于scan实现最终单值输出的逻辑，可以在scan链路后追加last()操作符取流的最后一个发射值，但这种写法没有reduce高效

额外注意事项 #

reduce/scan这类依赖流终止的聚合操作，要求上游的Observable是可以正常完成的（即流结束时会触发onComplete回调）；如果是永不终止的无限实时交易推送流，reduce会因为永远等不到流结束而不发射任何结果，这种场景需要搭配本地缓存当前累计状态的方案实现。

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