当前位置:首页 > 文章列表 > 文章 > 前端 > Web Streams 怎样处理背压和取消

Web Streams 怎样处理背压和取消

来源:17golang原创 2026-10-14 15:29:42 0浏览 收藏

Web Streams 处理“生产快、消费慢”时,正确做法是让管道自己传播背压,而不是在生产端无条件调用 enqueue()。把数据源放进 ReadableStream 的 pull(),让 TransformStream 只做转换,再用 pipeTo() 连接慢速 sink;取消时统一调用 AbortController.abort(),并在 source 的 cancel() 中关闭连接、定时器或文件句柄。

参考入口:https://developer.mozilla.org/en-US/docs/Web/API/Streams_API

背压解决“什么时候继续生产”,取消解决“什么时候彻底停止并释放资源”。前者看队列需求,后者看生命周期信号,两件事要同时设计。

先划清 source、transform 和 sink 的责任

可以把一条流想成三段:source 负责从网络、文件或业务生成数据;transform 负责解码、过滤或分片;sink 负责写入页面、缓存或另一个接口。队列属于流的内部状态,业务代码只应控制边界,不要把所有数据先收进数组。

一个实用的判断是:source 只回答“现在是否有需求”,transform 只回答“这个块如何变化”,sink 只回答“这个块何时真正写完”。职责混在一起时,最容易出现重复消费、无法停止和内存持续增长。

用 pull 与 desiredSize 让生产端尊重背压

可读流的队列会根据 queuing strategy 和 high water mark 计算需求。desiredSize 越小,表示可继续接收的空间越少;当它小于或等于零,生产端就不应继续无条件塞入新块。实际生产代码通常把读取动作放在 pull(),让运行时在消费者需要时再调用。

Web Streams source、队列、TransformStream 和慢速 sink 之间的背压结构说明图
图1:Web Streams 背压结构说明图,展示需求信号从消费端回传生产端。
function makeStream(getChunk) {
  let closed = false;

  return new ReadableStream({
    async pull(controller) {
      // pull 由下游需求触发;避免用无限循环提前填满队列。
      if (closed || controller.desiredSize 

这里的 desiredSize 是保护线,不是业务吞吐量指标。若每个块大小差异很大,应使用 ByteLengthQueuingStrategy 或自定义 size(),不要只按块数估算内存。

用 pipeThrough 和 pipeTo 连接转换与落点

pipeThrough() 把可读侧接到转换流,返回转换后的可读侧;pipeTo() 把最终结果交给 writable sink,并返回一个表示整条管道完成或失败的 Promise。这样,转换阶段不用自行维护“暂停/恢复”协议,队列压力会沿着管道传播。

const upperCase = new TransformStream({
  transform(chunk, controller) {
    // 转换只处理当前块;enqueue 后由下游队列继续施加压力。
    controller.enqueue(chunk.toUpperCase());
  },
});

const sink = new WritableStream({
  async write(chunk) {
    // 用异步写入模拟慢速落点;Promise 完成前不会确认本次写入。
    await saveChunk(chunk);
  },
  close() {
    // 正常结束时提交或关闭落点。
    console.log("sink closed");
  },
  abort(reason) {
    // 异常中止时撤销未完成的落点操作。
    console.warn("sink aborted:", reason);
  },
});

const pipePromise = source
  .pipeThrough(upperCase)
  .pipeTo(sink);

pipePromise.catch((error) => {
  // 统一处理 source、transform 或 sink 的失败。
  console.error("pipeline failed:", error);
});

不要在 sink 的 write() 里启动无法等待的后台任务,否则表面上写入已完成,实际队列仍在增长。要让背压有效,写入完成必须由返回的 Promise 表达。

用 AbortController 贯穿用户取消与超时

取消是“我已经不再需要这些数据”,不是“把流正常读完”。给 pipeTo() 传入 AbortSignal 后,页面离开、用户点击停止或超时都可以走同一条路径;source 的 cancel() 负责释放底层资源,sink 的 abort() 负责处理未完成写入。

AbortController 通过 pipeTo signal 触发 source cancel、sink abort 和资源清理的生命周期说明图
图2:取消与资源回收结构说明图,区分 cancel、abort 和正常 close。
const controller = new AbortController();
const timeoutId = setTimeout(() => {
  // 超时与用户主动停止共享同一个取消信号。
  controller.abort(new Error("stream timeout"));
}, 8000);

source
  .pipeThrough(upperCase)
  .pipeTo(sink, { signal: controller.signal })
  .then(() => {
    // 正常完成才走这里;取消通常会让 Promise rejected。
    console.log("pipeline completed");
  })
  .catch((error) => {
    // 区分取消和其他错误,避免把用户主动停止记成系统故障。
    if (controller.signal.aborted) {
      console.log("pipeline canceled:", controller.signal.reason);
    } else {
      console.error("pipeline failed:", error);
    }
  })
  .finally(() => {
    // 无论成功、失败还是取消,都不能遗留超时定时器。
    clearTimeout(timeoutId);
  });

stopButton.onclick = () => {
  // 用户不再需要结果时,停止整条 pipe。
  controller.abort("user stopped");
};

close() 表示正常结束,cancel() 表示可读侧放弃继续读取,abort() 表示 writable 侧被突然终止。三者都可能需要清理资源,但语义和错误传播不同,不能用一个“结束”回调替代。

按队列、关闭和取消边界做复查

上线前可以用下面这张速查表检查管道:

现象优先检查处理方向
内存随时间增长source 是否无条件 enqueue把读取放入 pull,检查 desiredSize 和 size 策略
取消后仍有新请求是否只取消了 UI 状态把 AbortSignal 传入 pipeTo,并实现 source.cancel
写入看似完成但落点滞后sink.write 是否返回真实 Promise等待底层写入结束再确认 chunk
错误被吞掉是否处理 pipeTo 返回的 Promise统一 catch,并记录 abort 与业务错误的区别

最后记住两个边界:队列高水位只限制“愿意缓存多少”,不能替代业务限流;AbortController 只发出取消信号,真正的连接关闭和资源释放仍要由 underlying source、transform 或 sink 实现。

常见问题

背压是不是只需要手动判断 desiredSize?

不一定。标准 pipe 链路会管理队列和传播需求,手动判断主要用于自定义 source 的生产策略;关键是不要绕开 pull 和写入 Promise。

调用 cancel 后还能继续读队列里的数据吗?

不能把它当作暂停。cancel 表示消费者放弃流,已经排队但未读的数据会被丢弃;如果只是暂时处理不过来,应让下游自然施加背压,而不是调用 cancel。

版本声明
本文转载于:17golang原创 如有侵犯,请联系study_golang@163.com删除
Vue 响应式对象解构后为什么失去更新Vue 响应式对象解构后为什么失去更新
上一篇
Vue 响应式对象解构后为什么失去更新
WebTransport 怎样选择双向流和 Datagram
下一篇
WebTransport 怎样选择双向流和 Datagram
查看更多
最新文章
查看更多
课程推荐
  • 前端进阶之JavaScript设计模式
    前端进阶之JavaScript设计模式
    设计模式是开发人员在软件开发过程中面临一般问题时的解决方案,代表了最佳的实践。本课程的主打内容包括JS常见设计模式以及具体应用场景,打造一站式知识长龙服务,适合有JS基础的同学学习。
    543次学习
  • GO语言核心编程课程
    GO语言核心编程课程
    本课程采用真实案例,全面具体可落地,从理论到实践,一步一步将GO核心编程技术、编程思想、底层实现融会贯通,使学习者贴近时代脉搏,做IT互联网时代的弄潮儿。
    516次学习
  • 简单聊聊mysql8与网络通信
    简单聊聊mysql8与网络通信
    如有问题加微信:Le-studyg;在课程中,我们将首先介绍MySQL8的新特性,包括性能优化、安全增强、新数据类型等,帮助学生快速熟悉MySQL8的最新功能。接着,我们将深入解析MySQL的网络通信机制,包括协议、连接管理、数据传输等,让
    500次学习
  • JavaScript正则表达式基础与实战
    JavaScript正则表达式基础与实战
    在任何一门编程语言中,正则表达式,都是一项重要的知识,它提供了高效的字符串匹配与捕获机制,可以极大的简化程序设计。
    487次学习
  • 从零制作响应式网站—Grid布局
    从零制作响应式网站—Grid布局
    本系列教程将展示从零制作一个假想的网络科技公司官网,分为导航,轮播,关于我们,成功案例,服务流程,团队介绍,数据部分,公司动态,底部信息等内容区块。网站整体采用CSSGrid布局,支持响应式,有流畅过渡和展现动画。
    485次学习
查看更多
AI推荐
  • PubMedQA数据集详解:生物医学问答基准、功能与应用指南
    PubMedQA
    深入了解PubMedQA生物医学问答数据集,涵盖其核心功能、使用方法及在临床决策、药物研发等场景的应用,助力提升NLP模型性能。
    471次使用
  • H2O EvalGPT:开源LLM大模型评估与排行榜工具
    H2O EvalGPT
    H2O EvalGPT是H2O.ai推出的开源LLM评估平台,提供详细的大模型性能排行榜、行业特定基准测试及A/B测试功能,助您快速选择最适合项目的高性能大语言模型。
    530次使用
  • LMArena是什么?伯克利AI模型评估平台使用指南与功能解析
    LMArena
    LMArena是加州大学伯克利分校推出的AI模型匿名评测平台。通过盲测投票机制,用户可对比不同大模型回答并生成实时排行榜,助力开发者优化模型及用户选择最佳AI工具。
    550次使用
  • 斯坦福HELM:大语言模型Holistic Evaluation整体评估框架详解
    HELM
    深入了解斯坦福推出的HELM(Holistic Evaluation of Language Models)大模型评测体系。本文解析其核心功能、安装配置步骤及应用场景,涵盖准确性、公平性、鲁棒性等多维度指标,助力开发者全面优化语言模型性能。
    501次使用
  • MMBench详解:多模态大模型基准测试、功能特点与使用指南
    MMBench
    MMBench是由上海人工智能实验室等机构联合推出的多模态基准测试平台,提供细粒度能力评估、大规模数据集及VLMEvalKit工具。本文详细介绍其核心功能、安装使用方法及应用场景,助力开发者全面评估多模态模型性能。
    318次使用