RxJS使用mergeMap/groupBy处理流后subscribe未触发问题

故障点是toArray()操作符的触发逻辑和你的业务场景完全不匹配：

• toArray()只会在上游流触发complete通知时，才会把期间收集到的所有值打包成数组向下游传递。
• 你用groupBy()生成的每个分组流是长期存活的：只要最上游的orders$（SignalRService里的普通Subject）没有调用complete()，分组流就会一直保持开放，持续等待后续新订单进入，永远不会触发complete，所以toArray()会一直卡着不输出值，后面的tap、subscribe自然拿不到数据。
• 硬编码of(orders)测试时逻辑正常，是因为of()在发完所有传入值后会立刻触发complete，刚好满足toArray()的执行条件。
• 第二次SignalR推送无响应也是同一个原因：第一次推送的订单进入分组流后，toArray()一直阻塞等待流结束，后续新进来的值根本不会被处理。

不要在长期存活的热点流上使用需要等待流完成才输出的操作符，根据你的业务场景（每次SignalR推送全量订单数组）选择对应实现即可：

方案1：数组层面直接分组（最适配当前场景，和手动分组逻辑完全一致） #

既然每次orders$推送的就是一整批完整订单数组，完全没必要把数组打平后做流级别的分组，直接在单批数据内做数组分组即可，逻辑简单性能更好：

constructor(private signalRService: SignalRService) {
  signalRService.orders$
    .pipe(
      map((orderList) => {
        // 初始化对应status 0/1/2的三个空分组
        const groups: OrderModel[][] = [[], [], []];
        orderList.forEach(order => {
          groups[order.status].push(order);
        });
        return groups;
      })
    )
    .subscribe(([group0, group1, group2]) => {
      this.orderCategory0.next(group0);
      this.orderCategory1.next(group1);
      this.orderCategory2.next(group2);
    });
}

方案2：保留groupBy写法，给分组加窗口边界 #

如果一定要用打平+groupBy的流处理写法，需要给分组逻辑增加批次边界，告诉RxJS每一批推送的数据处理完就结束当前分组窗口，不要永久等待：

import { windowToggle, mergeMap, groupBy, zip, of, toArray } from 'rxjs';

constructor(private signalRService: SignalRService) {
  signalRService.orders$
    .pipe(
      // 以每一次SignalR推送作为分组窗口的开启/关闭边界
      windowToggle(signalRService.orders$, () => signalRService.orders$),
      mergeMap(batchOrders => batchOrders.pipe(
        mergeMap(res => res),
        groupBy((order: OrderModel) => order.status),
        mergeMap(group => zip(of(group.key), group.pipe(toArray())))
      ))
    )
    .subscribe(([groupId, data]) => this.orders[groupId].next(data));
}

补充：如果后续SignalR调整为推送单条增量订单而非全量列表，groupBy的正确搭配是用scan做状态累积，不需要等流完成即可实时输出分组结果：

// 增量单条订单流场景示例
mergeMap(orderList => orderList),
groupBy(order => order.status),
mergeMap(group => group.pipe(
  scan((acc, cur) => [...acc, cur], [] as OrderModel[]),
  map(groupData => [group.key, groupData] as const)
))

• 对于长期存活的热点流（比如SignalR推送、全局事件流、WebSocket推送流），不要随意使用toArray()、reduce()这类必须等待上游complete才会输出的操作符，除非你明确通过窗口、取数等操作给流设置了结束边界。
• groupBy更适合处理零散、持续到来的单条数据的分组场景，如果每次拿到的都是全量数据集，直接在数组层面做分组的可维护性和性能都更优。

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