RxJs 合并含未完结流的Observable后拆分过滤流不生效问题

你代码中使用的group.pipe(toArray())是核心故障点：toArray()操作符会缓存上游所有发射的值，必须等上游流完全完结才会输出汇总后的数组。而你的dynamic$是基于WebSocket的长连接流，永远不会主动完结，因此toArray()始终无法触发输出，自然拿不到预期结果。

你的需求本质是每次合并流发射一整批用户数组时，直接在当前批次内按状态过滤即可，完全不需要将数组拆分为单个元素再用groupBy重组，简化后实现如下：

import { merge, BehaviorSubject, of, map, distinctUntilChanged, shareReplay } from 'rxjs';

// 原有类型和流定义保持不变
type User = { id: string; status: 'active' | 'inactive' };
const mockData: User[] = [
  { id: '1', status: 'active' },
  { id: '2', status: 'inactive' },
  { id: '3', status: 'active' }
];
const initial$: Observable<User[]> = of(mockData);
const sub$: BehaviorSubject<User[]> = new BehaviorSubject(mockData);
const dynamic$ = sub$.asObservable();

// 公共合并流，加shareReplay避免多订阅重复执行逻辑
const merged$ = merge(initial$, dynamic$).pipe(
  distinctUntilChanged((a, b) => JSON.stringify(a) === JSON.stringify(b)),
  shareReplay(1)
);

// 直接对每批次完整用户数组做过滤，无需拆包重组
const activeUsers$ = merged$.pipe(
  map(users => users.filter(u => u.status === 'active'))
);
const inactiveUsers$ = merged$.pipe(
  map(users => users.filter(u => u.status === 'inactive'))
);

// 测试用：模拟WebSocket推送新数据
setTimeout(() => {
  sub$.next([...mockData, {id: '4', status: 'active'}])
}, 1000)

• 放弃groupBy+toArray的冗余组合：你每次从合并流拿到的就是完整的用户数组，直接对数组做过滤逻辑更简单，性能也更好
• 新增shareReplay(1)：多个子流订阅合并流时，不会重复执行去重逻辑，也能保证新订阅直接拿到最新的用户数据
• 即时响应：每次合并流有新数据推送，两个过滤后的流会立刻输出对应状态的用户列表，不需要等待任何流完结，完全适配长连接WebSocket场景

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