如何使用RxJS的switchMap仅取消同一领域实体的待处理HTTP请求

这个问题我之前也碰到过，核心是要按实体ID做分组隔离，而不是用全局的switchMap一刀切——你现在的写法会让任何新请求都取消所有之前的待处理请求，不管它属于哪个实体。

解决方案思路 #

我们需要为每个实体ID维护一个独立的Subject流，每个流内部用switchMap来处理该ID的请求：当同一ID的新请求进来时，只会取消该ID下的旧请求，不同ID的流完全独立，互不干扰。

修改后的代码实现 #

import { of, Subject } from "rxjs";
import { delay, switchMap, tap } from "rxjs/operators";

// 存储每个实体ID对应的Subject
const entitySubjects = {};

// 模拟HTTP请求/计算的内部Observable
const innerCalculation = (x) => of("Calculation Result " + JSON.stringify(x)).pipe(
  tap(() => console.log('Begin Calculation', x)),
  delay(1000)
);

function startCalc(x) {
  const entityId = x.id;
  // 如果该ID还没有对应的Subject，创建并订阅它的处理流
  if (!entitySubjects[entityId]) {
    const subject = new Subject();
    entitySubjects[entityId] = subject;
    
    // 每个ID的流内部用switchMap，只取消当前ID的旧请求
    subject.pipe(
      switchMap((data) => innerCalculation(data))
    ).subscribe(result => console.log(result));
  }
  // 向对应ID的Subject发送新的计算请求
  entitySubjects[entityId].next(x);
}

// 测试调用
startCalc({id: 'A', data: 1}); // 实体A的第一次请求（会被后续A的请求取消）
startCalc({id: 'B', data: 1});
startCalc({id: 'A', data: 3}); // 实体A的第二次请求，取消第一次A的请求
startCalc({id: 'C', data: 1});
startCalc({id: 'D', data: 1});

运行结果说明 #

执行这段代码后，控制台会输出你预期的内容：

Begin Calculation { id: 'A', data: 1 }
Begin Calculation { id: 'B', data: 1 }
Begin Calculation { id: 'A', data: 3 }
Begin Calculation { id: 'C', data: 1 }
Begin Calculation { id: 'D', data: 1 }
Calculation Result {"id":"B","data":1}
Calculation Result {"id":"A","data":3}
Calculation Result {"id":"C","data":1}
Calculation Result {"id":"D","data":1}

可以看到：

• 实体A的第一次请求被第二次请求取消，所以不会输出它的结果
• B、C、D的请求都正常完成并输出结果
• 同一ID的请求只会互相影响，不同ID完全独立

额外优化（可选） #

如果实体ID不再被使用，你可以考虑清理对应的Subject（比如调用complete()并从对象中删除），避免内存泄漏。比如在计算完成后判断是否还有后续请求，或者监听实体的销毁事件来清理。

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