如何在RxJS中扫描对象流并新增cost累计求和属性

核心思路 #

使用RxJS内置的scan操作符完成状态累计，它会持续维护内部的累计状态，每次源流 emit 新值时都会触发计算并输出最新的结果，正好匹配需要输出每一步累计结果的需求。

scan操作符和数组的reduce逻辑类似，区别是reduce只会在源流 complete 后输出最终累计值，而scan会在每次收到源流新值时输出当前的累计结果，完全适配本需求场景。

完整实现代码 #

import { from } from 'rxjs';
import { scan } from 'rxjs/operators';

// 原始对象源流
const source$ = from([{ id: 1, cost: 10},{ id: 2, cost: 5},{ id: 3, cost: 3},{ id: 4, cost: 10}]);

// 派生流实现
const derived$ = source$.pipe(
  scan((acc, current) => {
    const totalCost = acc.totalCost + current.cost;
    return {
      ...current,
      totalCost
    };
  }, { totalCost: 0 }) // 初始化累计状态，初始总开销为0
);

// 测试订阅，输出结果和预期完全一致
derived$.subscribe(console.log);

输出结果 #

{ id: 1, cost: 10, totalCost: 10 }
{ id: 2, cost: 5, totalCost: 15 }
{ id: 3, cost: 3, totalCost: 18 }
{ id: 4, cost: 10, totalCost: 28 }

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