qinghuan
← 让长任务能暂停和恢复 / 全部笔记

单元 26 / 让长任务能暂停和恢复

P1 · 重点提升L2 · 进阶

流式与异步：用背压控制待处理数据

分清先显示一部分与真正完成，处理慢客户端、断线和取消，避免缓存越积越多。

清欢学习笔记概念解释 · 代码示例 · 动手练习
本篇目录 · 按需要跳转

读完这篇，你能做什么

能用模拟事件检查慢消费、异常和取消，限制缓冲大小，并正确处理结束状态和重复事件。

学习理由、准备要求与知识点

为什么学它

客户端读得慢或中途断线时，系统需要限制待处理数据，并让停止信号传下去，避免内存和后台任务持续增长。

先读这些内容

看懂 HTTP 请求、JSON 与 API 错误 ↗用列表、字典和函数处理第一份学习数据 ↗用状态机决定哪里需要 Agent ↗

知识点清单

  • 流式与异步的区别
  • 增量和完整结果
  • 网络分块与事件边界
  • SSE 事件标识
  • 有界队列和背压
  • TaskGroup 异常传播
  • 超时和取消传播
  • 断线重连与副作用边界

运行前先确认完整算法使用 Java 17 标准库；JDK 源码基线为 OpenJDK 17 GA。示例是内存中的机制模型，外部服务接入需要另外配置和验证。

01先弄懂它是什么

想让用户更早看到回答，可以逐步发送结果，这叫流式；等待某个操作时允许线程或任务先去做别的，是异步。两者可以组合，也可以分开：同步代码能逐块读取，异步请求也可能最后一次性返回全部内容。设计时先明确何时展示增量、何时算完整完成，以及用户取消后怎样让后续工作停下来。

网络每次传来的一块数据，不一定刚好是一条完整消息，SSE 的一个事件也未必就是整个业务结果。文本或 JSON 可能被拆到多个块里，工具参数必须等协议确认收齐后才能验证和执行。例如只收到一半参数就运行工具，会把原本的流式读取问题变成真实的错误操作。

如果后台生成得快、页面读取得慢，中间就会不断积压。背压是把“我暂时接不动了”的信号传回上游，让它按约定放慢或等待；另一种选择是明确规定哪些内容可以丢弃。有界队列只限制最多暂存多少条，装满后仍要选择阻塞、拒绝、超时或合并。队列设了上限，不等于上游已经会减速；使用无界队列则会把速度差转成持续增长的内存占用，不能靠它无限容纳慢客户端。

文中这几个词，先认一认

背压
下游处理慢时，让上游减速或限制继续接收数据，避免待处理内容在内存里无限堆积。
增量与最终结果
增量可以先供用户阅读；最终结果要等正常结束、收齐内容并通过约定校验后，才能正式提交。
取消传播
上层决定停止后，继续向下游任务和资源发送停止信号。能否立即停止，要看具体操作是否支持。

还有陌生词？到术语小抄里看一个具体例子 ↗

02看看它怎么工作

本题用固定容量环形缓冲演示有界队列：head指向下一读取位置，size记录元素数，写入位置为(head+size)对容量取模。队满offer返回false，队空poll返回空Optional；不覆盖尚未消费的数据。

容量三时依次写入一、二、三后队满，读取一使head前移，再写四会绕回原数组开头。逻辑顺序仍是二、三、四，数组物理排列可以不同。单独维护size避免仅凭head与tail相等无法区分空与满。

程序是单线程机制模型，不含锁或原子变量，不能跨线程直接共享。真实ArrayBlockingQueue通过同步和条件等待实现阻塞行为；应用仍需决定满队列政策、取消资源和最终结果状态。缓冲有界并不证明整个链路都有背压。

正在加载流程图…

图中状态只描述本篇可执行机制；业务调用仍需由外层应用负责。
查看流程图源码（Mermaid）
flowchart LR
  P["完整增量事件生产"] --> Q["有界队列"]
  Q --> C["慢消费者"]
  Q --> F["满时阻塞 拒绝 或合并"]
  F --> P
  C --> T["最终结果校验"]
  C --> X["取消向上游传播"]

按顺序看，每一步做了什么

  1. 定义增量事件和最终完成事件，保持协议消息边界。
  2. 使用有界缓冲并明确满队列时阻塞、拒绝或合并政策。
  3. 下游取消或超时向上游传播，释放连接与队列。
  4. 最终响应完整后执行结构和业务校验。
  5. 用慢消费者、断线和部分结果测试延迟与内存边界。

03对照实现代码

OpenJDK 17 ArrayBlockingQueue 源码 （新标签页）↗

OpenJDK 17 GA，jdk-17+35 · ArrayBlockingQueue.offer / put / take

下方为原创 Java 教学片段，调用已核验的 JDK API 说明局部机制；不是该类原样源码，也不实现神经网络。完整程序见算法模块。

核心机制代码 · java
ArrayBlockingQueue<Event> queue = new ArrayBlockingQueue<>(64);
boolean accepted = queue.offer(event); // 满时立即返回false，调用方必须处理。
if (!accepted) rejectOrCoalesce(event); // 应用政策，不是队列自动帮忙丢弃。
// put会等待空间，take会等待元素；选择阻塞接口时要处理线程资源与中断。

ArrayBlockingQueue 用固定大小的数组存放待处理数据，再用锁和条件变量协调放入与取出。队列满了时，offer 会返回 false，put 可以等待出现空位；队列空了时，take 可以等待新元素到来。选择方法前先确认是否允许等待，不能把这三个调用当作只有名称不同。

线程安全队列只保护队列内部，不自动让元素对象不可变，也不替应用处理慢客户端、网络取消或最终结果提交。放入可变DTO后继续修改仍可能产生竞争。

完整算法用单线程环形缓冲展示索引不变量，没有复制JDK并发机制。它不能因为数据结构相似就被宣传为线程安全替代品。

流式工具参数和普通文本事件的消费规则不同。队列可以传递两者，但是否完整、能否执行必须由协议状态与业务验证决定。

代码先调用谁，再调用谁

  1. 生产者得到完整增量事件
  2. offer尝试进入有界缓冲
  3. 满队列执行明确政策
  4. 消费者按FIFO处理
  5. 完成或取消时释放任务资源

04动手写一个例子

有界环形缓冲与回绕

固定正容量整数队列，offer在满时返回false且不覆盖，poll在空时返回空OptionalInt。仅单线程使用。Java17 Main.java覆盖满、空与回绕。

输入 1
容量2，写1/2/3
输出 1
true,true,false
输入 2
读1后写3，再依次读
输出 2
2,3

不需要每次读取都移动数组，head前进即可。写入位置由当前逻辑长度推导，取模让尾部回绕到数组开头。

size始终在零与容量之间，空和满用它直接判断。成功写入才增加size，成功读取才减少size，拒绝操作保持所有状态不变。

队列顺序通过逻辑位置定义：第j个元素在(head+j)模容量位置。即使数组内部看起来乱序，按这个规则读取仍保持FIFO。

为什么这样写是对的0<=size<=capacity；逻辑第j项位于(head+j)%capacity，offer只写空闲尾位，poll只取当前head，失败操作不修改队列。

时间 offer与poll均O(1)额外空间 O(capacity)

每次仅访问常数索引和计数，固定数组容量决定空间；整数元素大小固定。线程同步与网络缓冲不在本单线程算法中。

保存为 Main.java，运行 javac --release 17 -encoding UTF-8 Main.java，再运行 java Main。程序包含固定输入自测，断言失败时会抛出错误。

Main.java · 完整可运行程序 · Java
import java.util.*;
public class Main {
    static class Ring {
        final int[] values;int head=0,size=0;
        Ring(int capacity){if(capacity<=0)throw new IllegalArgumentException("capacity");values=new int[capacity];}
        boolean offer(int x){
            if(size==values.length)return false; // 满时不覆盖未消费元素。
            int tail=(int)(((long)head+size)%values.length);values[tail]=x;size++;return true; // 长整数相加避免索引和溢出。
        }
        OptionalInt poll(){
            if(size==0)return OptionalInt.empty();
            int x=values[head];head=(head+1)%values.length;size--;return OptionalInt.of(x); // 逻辑头前进并回绕。
        }
    }
    static void ok(boolean x){if(!x)throw new AssertionError();}
    static void eq(int e,OptionalInt a){if(a.isEmpty()||a.getAsInt()!=e)throw new AssertionError(a);}
    public static void main(String[] args){
        Ring r=new Ring(2);ok(r.poll().isEmpty());
        ok(r.offer(1));ok(r.offer(2));ok(!r.offer(3));
        eq(1,r.poll());ok(r.offer(3));eq(2,r.poll());eq(3,r.poll());ok(r.poll().isEmpty());
        Ring one=new Ring(1);ok(one.offer(Integer.MIN_VALUE));eq(Integer.MIN_VALUE,one.poll());ok(one.offer(Integer.MAX_VALUE));eq(Integer.MAX_VALUE,one.poll());
        try{new Ring(0);throw new AssertionError();}catch(IllegalArgumentException expected){}
        System.out.println("PASS 5 scenario groups");
    }
}

这些特殊情况也要试一试

  • 容量一也能区分空满并回绕。
  • 队满拒绝不能改变已有数据顺序。
  • 该实现没有同步，不可直接用于并发生产消费。
进阶选读 · 后续再读：跨阶段背压与结构化并发

场景模型流、重排和浏览器消费速度不同，取消时多个子任务仍运行。

做法给每段明确容量与取消归属，使用结构化并发或受控任务组聚合失败；合并文本增量时保留顺序和完整终止信号。

复杂度单队列操作可常数级，但系统排队延迟取决于到达率与服务率；不能由O(1)推出端到端低延迟或稳定。

好处与代价阻塞节流可能占线程，拒绝需要用户体验设计，丢弃可能破坏协议；按事件类型选择政策。

05项目里怎么用

将可合并展示增量与不可丢弃控制事件分开处理。文字片段可以在有明确顺序时合并，工具调用完成、错误和最终状态不能随意丢掉。若缓冲满导致控制事件丢失，前端可能永久显示运行中或错误执行半份结果。

取消传播到HTTP流、异步任务和下游处理，清理监听器与缓冲。某个远程副作用可能不支持取消，页面停止显示后仍需记录已发操作与结果未知。断线重连依赖事件ID和服务端保留策略，不能仅凭SSE支持就承诺任意历史可恢复。

性能测试故意放慢消费者，观察队列峰值、拒绝数、首输出与总耗时，保留失败样本。首字快但总任务很慢不一定改善体验；无界队列短时间测试正常也不能证明长期不会积压。

容易踩的坑

  • 把流式与异步视为同一个概念。
  • 收到一块工具参数就立即执行。
  • 使用无界队列掩盖长期速度不匹配。
  • 单线程环形缓冲被当成线程安全实现。

几种做法，怎么选

表格有四列，窄屏可左右滑动查看代价和适用条件。

方案优势代价与限制适用条件
阻塞等待可向生产者传递压力占用等待资源，需处理中断受控线程池与短等待
立即拒绝内存边界明确，失败及时调用方需处理重试或降级过载保护
增量合并减少展示事件开销必须保留顺序与控制事件可合并文本片段

06检验是否读懂了

先自己想、自己写，再展开答案核对。场景题是假设的练习情境；遇到还没学过的内容，可以做完基础后再回来。

先用自己的话解释

01有界队列自动保证背压吗？

展开参考答案与推导

不完全。队列限制缓冲，满时调用方必须执行阻塞、拒绝等政策；若继续在别处无限缓存，链路仍失控。

02head等于tail为何可能有歧义？

展开参考答案与推导

环形索引回绕后空与满都可能出现相同位置，需size或保留空槽等额外约定区分。

03流式增量能直接作为最终DTO吗？

展开参考答案与推导

不能，需等待完整终止并通过解析与业务验证，半份JSON或工具参数不能提交。

换个情况，还会处理吗

01浏览器慢但模型持续输出，如何防积压？

展开参考答案与推导

限定队列容量并按事件类型阻塞、合并或拒绝，取消时停止上游；同时测峰值与丢弃/拒绝状态。

02断线重连后重复执行工具怎么办？

展开参考答案与推导

工具执行依据完整调用ID与业务幂等记录，展示重放不应重新触发动作；区分事件回放和业务执行。

03队列线程安全但结果仍乱，查哪里？

展开参考答案与推导

检查元素是否在入队后被修改、多个生产者的业务排序以及终止事件处理；队列只保护自身操作。

写代码试一试

01满队列拒绝后内容如何？

输入 / 条件
容量2已有1,2，再offer3
输出 / 预期
仍是1,2

提示 失败操作不改状态。

展开参考答案与推导

先检查size，再写数组；若先覆盖尾位再返回false，会破坏未消费数据。

02怎样验证回绕？

输入 / 条件
写1,2，读1，写3
输出 / 预期
依次读2,3

提示 物理位置与逻辑顺序分开。

展开参考答案与推导

head移动后新tail回到空出的数组位置，poll按head读取而非从数组下标零重新遍历。

03为何需要并发版而非加volatile就够？

输入 / 条件
两个生产者同时offer
输出 / 预期
可能写同一位置

提示 复合更新需协调。

展开参考答案与推导

检查容量、计算尾位、写入和增加size是复合操作；仅可见性不能提供原子性，应使用成熟并发队列或完整同步设计。

读完之后，独立完成

课后作业：自己做一遍

本篇共 10 道作业，先完成 8 道基础与综合任务，再按前置知识挑战进阶题。其中包含 1 道思维拓展，试着换个条件，看看原来的办法是否还管用。不附参考答案。

这次练什么

能用模拟事件检查慢消费、异常和取消，限制缓冲大小，并正确处理结束状态和重复事件。

本篇重点加练

事件边界与取消传播共同影响流式结果完整性，重点检查取消时仍在途的数据。

  • 网络分块与事件边界
  • 超时和取消传播

重点内容会多练几次。先说清楚怎么做，再试试空数据、出错或条件改变时会发生什么。

先自己做，卡住时记下问题，再回来查。问答题写清“为什么、什么时候适用”；动手题留下输入、输出和运行步骤，方便下次重做。

基础必做

  1. 作业 01原理问答

    画出流式响应与异步任务两种交互过程，标记首个增量、完整结果和最终状态，解释客户端何时可以提交业务数据。

    提交内容
    一份独立书面回答或标注图，包含自造例子、概念边界及判断依据。
    怎样算完成
    • 逐项回应题目要求，区分事实、假设与个人判断。
    • 能用自己的例子解释适用条件，记录尚未弄懂的点。

    练到的知识点流式与异步的区别 · 增量和完整结果

  2. 作业 02代码推演

    把两个SSE事件拆成任意三段网络字节，再让一个事件跨中文字符边界；手工推演缓冲区、事件ID与解析时点。

    提交内容
    手工推演记录，写明输入约定、中间状态、最终判断；可运行时附程序对照记录。
    怎样算完成
    • 推演包含中间过程，数据形状、单位、顺序或状态前后一致。
    • 主动检查至少一个边界或异常条件，能解释手算与运行差异。

    练到的知识点网络分块与事件边界 · SSE 事件标识

  3. 作业 03原理问答

    定义用户取消时客户端缓冲区、已解析事件和在途字节的处理合同，说明哪个阶段之后数据不得再触发业务提交。

    提交内容
    概念或合同说明、关系图与自造案例，明确事实、假设和适用条件。
    怎样算完成
    • 关键概念有明确边界，能用案例检验声明。
    • 未获证据支持的假设单独标注。

    练到的知识点网络分块与事件边界 · 超时和取消传播

综合应用

  1. 作业 04面试场景

    生产端每秒100项、消费端每秒10项且队列上限20。设计队列满时策略，并说明TaskGroup中一个子任务失败后的行为契约。

    提交内容
    场景分析与决策记录，包含信息缺口、约束、备选方案和验证计划。
    怎样算完成
    • 方案回应给定约束，列出至少一种替代方案及代价。
    • 每项关键结论说明所需证据，不能核实的部分明确标注。

    练到的知识点有界队列和背压 · TaskGroup 异常传播

  2. 作业 05编程实践

    实现本地假流读取器，模拟中途取消、连接断开和重连重复事件；记录取消传播、资源释放与副作用执行次数。

    提交内容
    可复现的最小程序或实验材料、自写测试、环境说明与运行记录；题目允许纸面实验时提交完整案例表。
    怎样算完成
    • 提交物按说明可以重现；测试覆盖正常、边界与失败场景。
    • 运行证据与断言能核验声明的行为，注明模拟与真实执行范围。

    练到的知识点超时和取消传播 · 断线重连与副作用边界

  3. 作业 06排错实验

    故意把每个网络分块当作完整JSON解析，构造稳定复现输入；再检查慢消费者下内存与积压是否有界。

    提交内容
    最小故障输入、复现步骤、观测记录、定位依据及回归用例。
    怎样算完成
    • 其他人可按记录复现故障，并区分现象与已验证原因。
    • 定位过程有观测依据，回归用例可以再次暴露原问题。

    练到的知识点增量和完整结果 · 网络分块与事件边界 · 有界队列和背压

  4. 作业 07编程实践

    编写离线字节流回放器，在事件开始、中间和结束后分别注入取消，记录缓冲区、事件计数与清理行为，使用包含多字节字符的输入。

    提交内容
    可复现的小程序或题目允许的离线实验、自写测试、输入和运行记录。
    怎样算完成
    • 输入、环境和运行过程可重现，测试覆盖声明的重点条件。
    • 验证目标由合同确定，模拟结果不冒充真实服务证据。

    练到的知识点网络分块与事件边界 · 超时和取消传播

  5. 作业 08排错实验

    让取消事件与最后一个完成事件以两种顺序到达，检验客户端是否会双重结束或继续提交，保留最小可复现序列。

    提交内容
    最小故障序列、观测记录、原因核对与可再次触发原问题的测试。
    怎样算完成
    • 故障可以稳定重现，现象与已验证原因分开。
    • 观测记录足以判断声明的合同是否被违反。

    练到的知识点网络分块与事件边界 · 超时和取消传播

进阶挑战 · 掌握前置后选做

  1. 作业 09设计评审

    设计可恢复异步任务的流式观察接口，比较事件重放与业务动作重做的边界，制定客户端断线后的验收用例。

    提交内容
    设计评审文档或对照实验报告，包含前置条件、方案取舍、验收证据与尚未验证的范围。
    怎样算完成
    • 比较使用一致条件与事先声明的标准，记录失败与未知结果。
    • 说明方案的适用前提、未覆盖风险及后续验证方法。

    练到的知识点流式与异步的区别 · SSE 事件标识 · 超时和取消传播

思维拓展

先写出你的猜想，再用例子或实验验证。答案可以不同，但要说清依据、适用情况，以及还有什么没弄懂。

  1. 作业 10思维拓展 · 跨域迁移

    将流式边界与取消合同迁移到大文件分段上传的纸面场景，区分已接收字节、已校验分块与已提交文件之间的状态。

    提交内容
    原场景与目标场景的对应表、可迁移的机制、不能迁移的前提及验证设计。
    怎样算完成
    • 逐项核对两种场景的前提，不只比较相似术语。
    • 至少一个不能直接迁移的边界有具体检验方法。

    练到的知识点网络分块与事件边界 · 超时和取消传播

完成后，再检验一次

给每题标记“独立完成 / 查阅后完成 / 尚未完成”。查过资料或没做出的题，记下卡在哪里。隔一天换一组输入再试，能解释结果和出错原因，再把这个知识点标为掌握。

继续阅读

先看能帮你完成当前练习的部分。例子跑通后，再读更深入的解释。

打开资料库，找下一份学习材料 ↗