RxJS中如何实现按唯一名称重放各最新消息的通用可扩展方案？

问题背景 #

现有代码使用单个ReplaySubject(1)，订阅时仅能收到流中最后一次发送的消息（无论name字段），实际输出仅包含bbb 5的日志。但需求是：订阅时自动获取每个唯一name对应的最新消息（如aaa 3和bbb 5），且无需为100+个名称手动创建对应数量的ReplaySubject，需要通用可扩展的解决方案。

解决方案：基于RxJS操作符自动分组缓存 #

通过groupBy+shareReplay组合操作，自动按name字段分组并缓存每个分组的最新消息，新订阅时会自动获取对应分组的历史最新值，同时接收后续更新。

完整代码示例 #

// 创建原始Subject（无需使用ReplaySubject，分组后会单独缓存）
const source = new rxjs.Subject();

// 发送测试消息
source.next({name: 'aaa', payload: 1})
source.next({name: 'bbb', payload: 2})
source.next({name: 'aaa', payload: 3})
source.next({name: 'bbb', payload: 4})
source.next({name: 'bbb', payload: 5})

// 构建按名称分组并缓存最新值的处理流
const keyedCachedSource = source.pipe(
  // 按name字段拆分出独立分组流
  rxjs.groupBy(item => item.name),
  // 为每个分组流缓存最新1条数据，且共享给多订阅者
  rxjs.mergeMap(group => group.pipe(rxjs.shareReplay(1))),
  // 共享整个处理流，避免重复执行分组逻辑
  rxjs.share()
);

// 订阅aaa的消息（会立即收到最新的aaa 3，后续新消息也会触发）
keyedCachedSource.pipe(
  rxjs.filter(x => x.name === 'aaa')
).subscribe((x) => {
  console.log('I want this to be logged once ... received the latest aaa ' + x.payload)
});

// 订阅bbb的消息（会立即收到最新的bbb 5，后续新消息也会触发）
keyedCachedSource.pipe(
  rxjs.filter(x => x.name === 'bbb')
).subscribe((x) => {
  console.log('I want this to be logged once ... received the latest bbb ' + x.payload)
});

// 测试后续消息发送
source.next({name: 'aaa', payload: 6});
// 输出：I want this to be logged once ... received the latest aaa 6

核心逻辑说明 #

• groupBy(item => item.name)：将原始数据流拆分为多个子流，每个子流对应唯一的name值，自动管理所有名称分组。
• shareReplay(1)：为每个分组流缓存最新的1条消息，新订阅该分组时会立即推送缓存的最新值，实现类似ReplaySubject(1)的效果。
• mergeMap：将所有分组流合并为一个统一的数据流，保持原有的消息推送顺序，方便后续过滤订阅。
• share()：共享处理后的数据流，避免每次订阅都重复执行分组和缓存逻辑，提升性能并保证状态一致性。

这种方案无需手动维护多个Subject，无论新增多少个名称都能自动适配，完全满足通用、可扩展的需求。

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