如何使用RxJS实现支持重复请求取消的可取消HTTP队列

需求规则 #

队列需要满足以下执行逻辑：

• 点击Item A「加载更多」，发起Item A的HTTP请求
• 点击Item B「加载更多」，发起Item B的HTTP请求
• Item B请求pending时再次点击Item B「加载更多」：自动取消上一次未完成的Item B请求，重新发起新的Item B请求
• Item B请求pending时点击Item A「加载更多」：不做拦截，直接并行发起Item A的HTTP请求

原有实现的问题 #

最初用takeUntil+concatMap的方案有两个核心缺陷：

• concatMap会串行处理所有队列请求，不同Item的请求会互相阻塞，不符合不同Item可并行的要求
• 全局的sameItem$无法精准匹配当前执行中请求的标识，容易误取消其他Item的请求
  动态选择switchMap/concatMap的思路方向正确，但不需要手动判断队列元素，用RxJS内置的分组操作符可以更简洁实现。

实现代码 #

核心用groupBy+mergeMap+switchMap三个操作符组合，不需要额外维护取消触发器：

• groupBy：按Item唯一标识（比如item.id）对流做分组，同一个Item的请求会进入同一个分组流
• mergeMap：并行处理所有分组流，保证不同Item的请求不会互相阻塞
• switchMap：每个分组内部用switchMap处理，同分组新请求进来时自动取消上一次未完成的请求，Angular HttpClient会在流取消时自动中断底层HTTP请求

Service层实现 #

import { Injectable } from '@angular/core';
import { Subject, mergeMap, groupBy, switchMap, map } from 'rxjs';
import { HttpClient } from '@angular/common/http';

// 队列元素类型，必须带唯一标识id
interface QueueItem {
  id: string;
  url: string;
}

@Injectable({ providedIn: 'root' })
export class RequestQueueService {
  private readonly queue$ = new Subject<QueueItem>();

  constructor(private http: HttpClient) {
    this.queue$.pipe(
      groupBy(item => item.id),
      mergeMap(groupedStream$ => 
        groupedStream$.pipe(
          switchMap(item => this.http.get(item.url).pipe(
            // 把响应和item id绑定，方便后续分发处理
            map(response => ({ itemId: item.id, data: response }))
          ))
        )
      )
    ).subscribe(({itemId, data}) => {
      console.log(`Item ${itemId} 请求响应`, data);
      // 这里可以加响应分发逻辑，比如通过Subject通知对应组件
    });
  }

  addRequest(item: QueueItem) {
    this.queue$.next(item);
  }
}

组件层调用 #

// 组件内调用
loadMore(item: QueueItem) {
  this.requestQueueService.addRequest(item);
}

这个实现完全匹配需求：不同Item的请求并行执行互不干扰，同一个Item重复触发时自动取消上一次未完成请求，没有多余的网络开销，也不需要手动维护AbortController或者取消信号。

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