Angular RxJS使用share、map、filter时数据更新异常问题咨询

• _allUsers$ 使用了share()操作符，会将上游Firestore返回的全量用户数据做缓存，作为全局共享的热Observable，新订阅接入时会优先推送最近一次缓存的旧数据，不会等待Firestore返回最新的全量用户列表。
• 组件中调用getTalents()时添加了take(1)操作符，只要流吐出第一个值（也就是share()缓存的旧用户数据）就会立刻终止订阅，即使后续Firestore返回了最新的用户数据，也不会触发重新过滤逻辑，最终得到的就是旧数据基于新搜索条件过滤的结果，甚至是完全过期的旧列表。
• 若组件销毁时没有主动调用this.subs.unsubscribe()清理订阅，还会存在内存泄漏，旧订阅的残留逻辑会进一步干扰新搜索的结果。

根据你的业务场景可以选择以下任意一种方案：

方案1：去掉全局共享流，每次搜索发起独立查询（推荐，逻辑最简单） #

如果不需要全局复用全量用户数据，直接去掉全局共享的_allUsers$定义，每次调用getTalents时发起独立的Firestore查询，天然避免缓存问题：

// 修改dataService中的getTalents方法
getTalents(searchCriteria) {
    // 每次查询都从Firestone拉取最新数据
    return firestore.collection<any>('users').valueChanges().pipe(
        tap((talents) => console.log(talents)),
        map((talents: any) => {
            let filtered = talents.filter(
                (t) =>
                    (t.userType === 'talent' || t.userType === 'both') &&
                    this.matchOccupationOrSkill(
                        searchCriteria.occupation,
                        t.occupation,
                        t.skills
                    )
            );
            if (searchCriteria.loggedInUserId) {
                filtered = filtered.filter(
                    (t) => t.id !== searchCriteria.loggedInUserId
                );
            }
            return filtered;
        }),
        take(1) // 此时添加take(1)不会有问题，拿到最新数据就终止流
    );
}

方案2：保留共享流，去掉take(1)接收实时更新 #

如果需要全局缓存用户数据减少Firestore请求，就去掉订阅时的take(1)，让流可以持续接收最新的用户数据更新，同时务必添加组件销毁时的订阅清理逻辑：

// 修改TalentSearchResult组件代码
ngOnInit(): void {
    this.subs.add(
        this.route.queryParams.subscribe(params => {
            if (params.action==='homesearch' || params.action==='jobsearch'){
                this.criteria = {
                    occupation:params.occupation,
                    industry:params.industry,
                    loggedInUserId:params.loggedInUserId,
                }
                // 去掉take(1)，用户数据更新时会自动重新过滤
                this.subs.add(
                    this.dataService.getTalents(this.criteria).subscribe(talents => {
                        this.count = talents.length;
                        if (this.count >0){
                           this.Talents = talents.sort((p1,p2)=> p2.createdate-p1.createdate);
                        }
                    })
                )
            }
        })
    )
}

// 新增销毁逻辑清理所有订阅，避免内存泄漏
ngOnDestroy(): void {
    this.subs.unsubscribe();
}

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