使用NgRx Effect触发多次后端调用时请求被取消的问题

你遇到的前12次请求被取消的核心原因，是loadItemCollectionWithSkip$ Effect中使用了switchMap操作符。

switchMap的特性是：每当新的Observable（这里对应新的loadItemCollectionWithSkip动作）进入时，会自动取消之前仍在执行的Observable。你一次性派发了13个请求动作，switchMap只会保留最后一个请求的Observable，前面12个请求因此被强制中断。

下面是几种针对性的解决方案：

方案1：串行执行请求（替换为concatMap） #

如果不需要并发，想保持类似之前递归的串行执行逻辑，把switchMap换成concatMap即可。它会等待前一个请求完成后，再处理下一个动作对应的请求，不会取消任何请求：

loadItemCollectionWithSkip$ = createEffect(() =>
    this.actions$.pipe(
        ofType(this.loadActionGroup.loadItemCollectionWithSkip),
        concatMap((action) => { // 替换switchMap为concatMap
            return this.loadService.getListOdata(this.filterValue, action.skip).pipe(
                map((items: T[]) =>{
                    return this.loadActionGroup.loadItemSuccess({ items })
                })
            );
        })
    )
);

方案2：并行执行并控制并发数（替换为mergeMap） #

如果想提升速度，同时避免触发API的并发限制，可以用mergeMap并指定并发数。比如同时最多执行3个请求：

loadItemCollectionWithSkip$ = createEffect(() =>
    this.actions$.pipe(
        ofType(this.loadActionGroup.loadItemCollectionWithSkip),
        mergeMap((action) => { // 替换switchMap为mergeMap
            return this.loadService.getListOdata(this.filterValue, action.skip).pipe(
                map((items: T[]) =>{
                    return this.loadActionGroup.loadItemSuccess({ items })
                })
            );
        }, 3) // 第二个参数控制并发数，可根据API限制调整
    )
);

方案3：直接批量发起请求（用forkJoin） #

可以在获取总条数后，直接生成所有请求并批量执行，无需派发多个动作，逻辑更集中可控：

loadCollectionCount$ = createEffect(() =>
    this.actions$.pipe(
        ofType(this.loadActionGroup.loadItemCollection),
        withLatestFrom(this.store.select(this.loadedStatus)),
        switchMap(([{ }, loaded]) => {
            if (!loaded) {
                return this.loadService.getListCount(this.filterValue).pipe(
                    switchMap((result: number) => {
                        // 生成所有skip对应的请求Observable
                        const requests = [];
                        for (let i = 0; i < result; i += 500) {
                            requests.push(
                                this.loadService.getListOdata(this.filterValue, i).pipe(
                                    // 单个请求失败时返回空数组，避免影响整个批量请求
                                    catchError(() => of([]))
                                )
                            );
                        }
                        // 并行执行所有请求，等待全部完成后合并结果
                        return forkJoin(requests).pipe(
                            tap(() => this.store.dispatch(
                                this.loadActionGroup.loadItemCollectionCountSuccess({ count: result })
                            )),
                            map((allItems) => {
                                const mergedItems = allItems.flat();
                                return this.loadActionGroup.loadItemSuccess({ items: mergedItems });
                            }),
                            catchError(error => {
                                console.log(error, `${this.typeName} effects - loadCollection`);
                                return of(this.loadActionGroup.loadItemFailure({ error: { ...error } }))
                            })
                        );
                    }),
                    catchError(error => {
                        console.log(error, `${this.typeName} effects - loadCollection`);
                        return of(this.loadActionGroup.loadItemFailure({ error: { ...error } }))
                    })
                )
            } else {
                return of();
            }
        })
    )
);

如果需要串行执行批量请求，把forkJoin换成concat即可。

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