Angular RxJS中如何处理多数据流的同对象属性变更并复用订阅逻辑

算力阿峰

2026-4-29

用RxJS合并两个数据流，复用处理逻辑

刚接触RxJS确实容易被一堆操作符绕晕，你的需求其实用merge操作符就能轻松解决！

核心方案：使用`merge`合并流

merge的作用就是把多个Observable合并成一个新的Observable——只要任意一个源Observable发射了值，合并后的流就会立刻把这个值推送给订阅者，完美契合你要监听两个数据流变更、复用同一段代码的需求。

修改你的代码如下：

// 先导入需要的RxJS操作符
import { merge } from 'rxjs';
import { distinctUntilChanged } from 'rxjs/operators';

// 替换YourItemType为你实际的对象类型
displayItem$(selectedItems$: Observable<YourItemType>, updatedItem$: Observable<YourItemType>) {
  merge(selectedItems$, updatedItem$)
    // 可选优化：避免重复处理无变化的对象
    .pipe(
      // 自定义比较逻辑：判断前后对象是否真的有属性变化
      distinctUntilChanged((prev, curr) => {
        // 先通过唯一ID判断是否是同一个对象
        if (prev.id !== curr.id) return false;
        // 简单场景可以用JSON.stringify对比属性，复杂场景建议写具体属性的判断逻辑
        return JSON.stringify(prev) === JSON.stringify(curr);
      })
    )
    .subscribe(item => {
      this.displayLatest(item);
    });
}

为什么选`merge`而不是其他操作符？

concat：会按顺序处理流，必须等第一个流完成后才会处理第二个流，完全不适合你要实时监听两个流的场景。
combineLatest：需要所有流都至少发射一次值，之后每次有流发射新值时，会把所有流的最新值组合成数组返回——但你只需要单个最新的对象，不需要组合两个值，所以merge更贴合需求。