React Saga如何移除actionChannel队列中等待的指定action

Redux-Saga 原生的channel/actionChannel基于FIFO队列实现，没有提供从队列中间定向删除特定消息的API，因此无法直接操作channel内部存储移除待执行任务，我们可以通过「待执行任务标记+取任务时校验+运行中任务单独取消」的方案实现需求，同时覆盖「队列等待」和「正在执行」两种场景下的取消逻辑。

实现逻辑拆解 #

• 每个下载请求生成全局唯一的requestId，作为请求和取消操作的关联标识
• 维护两个映射表：
  • pendingTasks：存储所有已入队、还未被worker拾取执行的任务
  • runningTasks：存储所有已被worker拾取、正在执行doDownload的任务对应的saga task实例，用于中断正在运行的任务
• worker从channel取到任务后，先校验该任务是否还在pendingTasks中：如果已经被移除（被取消），直接跳过不执行下载逻辑
• 单独监听取消action，收到取消事件时：
  • 先从pendingTasks中删除对应任务，覆盖「任务还在队列等待」的场景
  • 如果对应任务存在于runningTasks中，调用cancel中断执行中的下载任务，覆盖「任务已经开始运行」的场景

完整实现代码 #

import { channel, take, fork, call, put, cancel } from 'redux-saga/effects';
import type { Channel, Task } from 'redux-saga';

// 类型定义可根据业务调整，核心要求payload携带唯一requestId
type requestPayload = {
  requestId: string;
  // 其余下载业务参数，比如文件地址、保存路径等
  [key: string]: unknown;
};

// 假设action常量定义：
// requested: 触发下载的action，结构为 { type: 'download/requested', payload: requestPayload }
// cancelRequest: 取消下载的action，结构为 { type: 'download/cancel', payload: { requestId: string } }

function* handleRequest(
  streamChannel: Channel<requestPayload>,
  pendingTasks: Map<string, requestPayload>,
  runningTasks: Map<string, Task>
) {
  while (true) {
    const payload = yield take(streamChannel);
    const { requestId } = payload;

    // 任务已被取消，直接跳过不执行
    if (!pendingTasks.has(requestId)) {
      continue;
    }

    // 从待执行列表移除，标记为进入执行状态
    pendingTasks.delete(requestId);

    // 启动下载任务，记录task实例用于后续取消
    const downloadTask = yield fork(function* () {
      try {
        yield call(doDownload, payload);
      } finally {
        // 任务执行完成/被取消，都从运行列表清理
        runningTasks.delete(requestId);
      }
    });
    runningTasks.set(requestId, downloadTask);
  }
}

function* watchRequest() {
  const chan: Channel<requestPayload> = yield call(channel);
  const pendingTasks = new Map<string, requestPayload>();
  const runningTasks = new Map<string, Task>();

  // 启动固定数量的worker
  for (let i = 0; i < WORKER_COUNT; i++) {
    yield fork(handleRequest, chan, pendingTasks, runningTasks);
  }

  while (true) {
    // 同时监听新增下载、取消下载两类action
    const action = yield take([requested, cancelRequest]);
    
    if (action.type === requested.type) {
      const { payload } = action;
      // 新任务先加入待执行映射表，再推入channel队列
      pendingTasks.set(payload.requestId, payload);
      yield put(chan, payload);
    }

    if (action.type === cancelRequest.type) {
      const { requestId } = action.payload;
      // 先从待执行列表删除，覆盖队列等待场景的取消
      pendingTasks.delete(requestId);
      // 如果任务正在运行，直接中断执行
      if (runningTasks.has(requestId)) {
        yield cancel(runningTasks.get(requestId));
        runningTasks.delete(requestId);
      }
    }
  }
}

场景验证（对应2个worker+3个请求的case） #

1. 同时派发3个requested action：前2个任务被空闲worker取走，从pendingTasks移入runningTasks开始执行doDownload；第3个任务存入pendingTasks，同时在channel队列中排队等待worker
2. 此时派发第3个任务对应的cancelRequest action：逻辑会直接将第3个任务从pendingTasks中移除，因为任务还未被worker拾取，不会操作runningTasks
3. 等后续有worker空闲，从channel中取到第3个任务的payload时，校验发现pendingTasks中已经没有对应requestId，直接跳过不执行，达到从队列移除取消任务的效果

注意事项 #

• 每个请求的requestId必须全局唯一，推荐用时间戳+随机数、uuid等方式生成，不要用数组下标等易重复的标识，避免取消错任务
• saga的cancel只会中断saga自身的执行流程，如果doDownload内部有原生异步逻辑（如fetch请求、文件流写入），需要自行在逻辑中处理中断（如fetch传入signal、调用stream.abort()），避免资源泄漏
• 被取消的队列消息虽然还留在channel的内部队列中，但worker取到后会直接跳过，额外开销可以忽略，不需要自行实现复杂的自定义channel来做物理删除，稳定性更高

内容的提问来源于stack exchange，提问作者Y.Yuksel