在Deno.serve流式响应中处理客户端断开/终止连接问题

要处理客户端中途断开连接的情况，你可以利用ReadableStream的cancel回调，结合AbortController来终止数据生成的worker任务，具体实现如下：

修改后的代码示例 #

Deno.serve((req) => {
  // 创建AbortController，作为控制worker终止的信号源
  const abortController = new AbortController();
  const { signal } = abortController;

  // 将终止信号传入迭代器，让worker能感知终止事件
  let asyncIterator = getChunkIterator(req.url, { signal });

  const stream = new ReadableStream({
    async pull(controller) {
      try {
        // 若信号已中止，直接关闭流控制器
        if (signal.aborted) {
          controller.close();
          return;
        }
        
        const { value, done } = await asyncIterator.next();
        if (done) {
          controller.close();
          return;
        }
        controller.enqueue(value);
      } catch (err) {
        // 捕获中止信号错误，避免无意义报错
        if (err.name === "AbortError") {
          controller.close();
        } else {
          controller.error(err);
        }
      }
    },
    // 客户端断开时，流会触发cancel回调
    cancel(reason) {
      // 触发中止信号，通知worker停止生成数据
      abortController.abort(reason);
    }
  });
  
  const response = new Response(stream);
  return response;
});

适配worker的迭代器实现 #

你需要调整getChunkIterator，让它能监听AbortSignal，在信号触发时停止worker并终止迭代：

async function* getChunkIterator(url, { signal }) {
  // 监听中止信号，触发时执行worker停止逻辑
  signal.addEventListener("abort", () => {
    // 这里添加停止对应worker的代码，比如终止子进程、清理资源等
    stopWorkerForUrl(url);
  });

  try {
    while (!signal.aborted) {
      // 生成数据块的业务逻辑
      const chunk = await generateChunkForUrl(url);
      yield chunk;
    }
  } catch (err) {
    // 忽略中止错误，避免向上抛出干扰流程
    if (err.name !== "AbortError") throw err;
  }
}

关键逻辑说明 #

• AbortController：作为响应流和worker之间的控制桥梁，客户端断开时通过abort()触发终止信号。
• ReadableStream的cancel回调：客户端主动断开连接时，Deno会触发这个回调，我们在这里触发中止信号通知worker停止。
• 迭代器监听信号：getChunkIterator通过监听signal的中止事件，执行worker停止逻辑，同时终止数据生成的循环。

这样就能在客户端中途断开时，及时停止对应的数据生成worker，避免不必要的资源消耗。

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