agent_backend/public/downloads/tasks-lessons.md

异步任务开发

术核 AI · 后端工程课件

01 提交导出任务，返回 202 与可查询状态

运营想导出两件商品的清单。导出需要等待，我们先让页面拿到一个编号，随时查看处理到了哪一步。

从一个问题开始

运营想导出两件商品的清单。导出需要等待，我们先让页面拿到一个编号，随时查看处理到了哪一步。

这次的终点

提交商品编号后得到 202 和查询地址，并用脚本验证任务只能按约定顺序走向完成。

开始前，确认你在这里

• 已完成 FastAPI 的请求参数与响应练习，能在 /docs 发出请求；在 shop-service 根目录新开终端，先确认当前路径，本课才创建 tasks_lab。
• 本课先创建 tasks_lab 及其独立 .venv；完成第 1 步后所有文件名都相对 tasks_lab。以后每个新终端先进入该目录并激活这里的 .venv，不使用 shop-service 根目录的 .venv。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 提交商品编号后得到 202、任务编号与 Location 查询地址
• 查询同一编号，区分已排队和已经完成的导出状态
• 用脚本验证 queued→running→succeeded，并拒绝从终态回退

一步步完成案例

1. 准备隔离练习目录

在 shop-service 下创建 tasks_lab 并进入，使用 Python 3.11+ 虚拟环境。第 01–07 课每个文件独立运行。Windows 可用 py -3 -m venv .venv 和 ..venv\Scripts\Activate.ps1；下面是 Linux/WSL2 命令。没有启动任何外部生产服务。

文件：终端 / 当前课时

mkdir tasks_lab
cd tasks_lab
python3 -m venv .venv
source .venv/bin/activate
python -m pip install "fastapi>=0.115,<1" "uvicorn[standard]>=0.30,<1" "httpx>=0.27,<1"
python --version

你应该看到什么

安装结束后 python --version 为 3.11 或以上，终端当前目录是 tasks_lab；尚未启动导出服务。

刚才发生了什么

这个目录把本模块文件与商品 API 分开；虚拟环境让安装的包只服务于这一组练习。后面看到的 python 命令都使用这里的解释器。

2. 定义状态机

创建 api_states.py。transition 只用于同进程实验，先检查再修改；错误时不能留下半更新记录。保留原始 product_ids，后续任务执行只接收业务标识或可序列化快照。

文件：api_states.py

from uuid import uuid4
from fastapi import FastAPI, HTTPException, Response
from pydantic import BaseModel, Field

app = FastAPI()
jobs: dict[str, dict] = {}
allowed = {
    "queued": {"running", "cancelled"},
    "running": {"succeeded", "failed", "cancelled"},
    "succeeded": set(), "failed": set(), "cancelled": set(),
}

class ExportIn(BaseModel):
    product_ids: list[int] = Field(min_length=1, max_length=100)

def transition(job_id: str, target: str):
    row = jobs[job_id]
    if target not in allowed[row["status"]]:
        raise ValueError(f'{row["status"]} -> {target} is not allowed')
    row["status"] = target

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。保存 api_states.py 后文件中有五种状态和 transition 函数；直接运行文件不会启动服务，也不会产生任务。

刚才发生了什么

queued 表示已排队，running 表示开始处理，后三种是完成、失败和取消。allowed 列出允许的下一步；先检查再赋值，可以避免错误转换改坏原记录，这种规则表叫状态机。

3. 增加提交与查询

把这段追加到同一文件。返回 202 和 Location，查询不存在的编号返回 404。当前所有任务保持 queued，不能通过等待假定它会完成。运行 python -m uvicorn api_states:app --reload 后在 /docs 操作。

文件：api_states.py

@app.post("/exports", status_code=202)
async def submit_export(body: ExportIn, response: Response):
    job_id = str(uuid4())
    jobs[job_id] = {
        "id": job_id, "status": "queued",
        "product_ids": body.product_ids, "result": None, "error": None,
    }
    response.headers["Location"] = f"/exports/{job_id}"
    return jobs[job_id]

@app.get("/exports/{job_id}")
async def get_export(job_id: str):
    if job_id not in jobs:
        raise HTTPException(404, "导出任务不存在")
    return jobs[job_id]

你应该看到什么

在 /docs 向 POST /exports 提交 {"product_ids":[1,2]}，得到 202、一个 id 和 Location；按地址查询得到同一编号及 queued。

刚才发生了什么

202 说明服务已受理，结果还没产生；Location 告诉调用者去哪里查。字典目前只存于这个服务进程，本课尚未接入执行函数，所以一直 queued 是正确现象。

4. 验证成功与非法转换

单独创建 verify_states.py，执行 python verify_states.py。TestClient 与 transition 在同一进程共享字典，这不是修改另一个已启动服务的状态。记录 202、Location、queued→running→succeeded，以及终态回退失败。

文件：verify_states.py

from fastapi.testclient import TestClient
from api_states import app, jobs, transition

with TestClient(app) as client:
    response = client.post("/exports", json={"product_ids": [1, 2]})
    assert response.status_code == 202
    job_id = response.json()["id"]
    assert response.headers["location"] == f"/exports/{job_id}"
    transition(job_id, "running")
    jobs[job_id]["result"] = {"rows": 2}
    transition(job_id, "succeeded")
    assert client.get(f"/exports/{job_id}").json()["status"] == "succeeded"
    try:
        transition(job_id, "running")
    except ValueError:
        print("terminal transition rejected")

你应该看到什么

python verify_states.py 最后打印 terminal transition rejected，其余 assert 不报错；查询返回 succeeded。

刚才发生了什么

TestClient 在脚本自己的进程里调用同一个应用，因此能直接推进它的内存状态。拒绝从 succeeded 回到 running 证明成功结果不会被随意覆盖；这不等于给另一个终端的服务完成了任务。

检查学习结果

• POST 返回 202，Location 能查询到同一 job_id。
• 新提交任务为 queued，未知 job_id 返回 404。
• succeeded→running 被拒绝，原状态仍为 succeeded。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

用户导出 100 条商品，希望关闭页面后仍能查询结果。

实现决策

• job_id 是独立资源标识，不能用请求连接标识代替。
• 正式受理需要持久化输入与任务记录，本课内存接口不能承担该承诺。
• 结果保存引用和校验信息，不把大文件塞进查询响应。

隔离环境演练

只在本地提交一条 queued 任务后重启服务，记录查询变成404，说明进程内存的丢失边界。

恢复验收

• 接口能区分受理、运行与终态。
• 非法终态回退不能覆盖结果。
• 文档明确本课没有自动执行和持久化。

遇到问题时查这里

客户端把 202 当成下载成功

原因：受理与执行完成混在一个判断中。

处理：仅在查询得到 succeeded 且存在结果引用后开放下载。

重启后已受理任务全部消失

原因：状态只存在内存，尚未连接可靠执行系统。

处理：将本课限定为协议实验，后续使用持久状态和 outbox。

自己试一次

增加 GET 查询中的 terminal 布尔字段，验证 queued 为 false、succeeded 为 true。

提示：根据状态计算字段，不另存一份可能不同步的布尔值。

参考答案：

返回记录时增加 terminal=row['status'] in {'succeeded','failed','cancelled'}；不要让客户端传入 terminal。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

202 的含义

完成第 3 步“增加提交与查询”后，再回顾这条规则：202 只说明服务受理了请求，不能表示文件已经产生。提交响应应提供 job_id、当前状态和查询地址；最终结果通过查询接口获取。

状态转换是业务规则

完成第 2 步“定义状态机”后，再回顾这条规则：本模块使用 queued、running、succeeded、failed、cancelled。running 才能进入成功或失败；终态不再被旧执行者改写。取消请求与已取消会在后续拆开处理。

先使用受控的内存实验

完成第 4 步“验证成功与非法转换”后，再回顾这条规则：本课只建立协议，不启动后台执行。记录存在当前进程内，重启会丢失。第 11 课才把状态迁入 MySQL；不能把本课 202 作为可靠交付承诺。

技术文档

• HTTP 202
• FastAPI 状态码

---

02 用 TaskGroup、队列与超时约束 asyncio

导出前需要读取 20 件商品。逐件等太慢，一次启动太多又会把等待任务堆满内存；本课用可控的模拟读取观察两者。

从一个问题开始

导出前需要读取 20 件商品。逐件等太慢，一次启动太多又会把等待任务堆满内存；本课用可控的模拟读取观察两者。

这次的终点

正常批次收集到 20 条；缩短时间或让单条失败后，整批停止并输出消费者清理记录。

开始前，确认你在这里

• 已完成本模块第 01 课；在 tasks_lab 的虚拟环境终端运行独立脚本，确认 python --version ≥ 3.11。无需运行 API 或连接数据库。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 用四个消费者、容量四的队列和两张通行证限制批量读取
• 验证正常预算内得到 20 件商品，缩短预算后整批超时
• 让单条读取失败，观察其他消费者取消并完成清理

一步步完成案例

1. 定义可等待的读取函数

新建 bounded_async.py，使用 sleep 代替商品 API 的 I/O。把输出保存在字典，以 product_id 建立对应关系；实际外部调用还必须设置连接和读取超时。

文件：bounded_async.py

import asyncio

async def read_product(product_id: int):
    await asyncio.sleep(0.1)
    return {"id": product_id, "name": f"商品-{product_id}", "description": "教学样品", "is_active": False, "price_cents": 0}

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。保存函数后可看到每次等待 0.1 秒并返回当前编号对应商品；此时没有入口，不会自动打印 20 条。

刚才发生了什么

async def 声明一段可暂停的工作，await sleep 把等待时间让给别的工作。这里模拟网络等候，不会真的读取商品表。

2. 建立固定消费者与有限缓冲

追加函数。四个消费者最多有两个同时读取；队列最多保存四项。None 是本实验的结束哨兵，真实输入不使用 None。task_done 必须放在 finally，失败也要结清已取出的项。

文件：bounded_async.py

async def collect_products(product_ids):
    queue = asyncio.Queue(maxsize=4)
    slots = asyncio.Semaphore(2)
    results = {}

    async def consume():
        try:
            while True:
                product_id = await queue.get()
                try:
                    if product_id is None:
                        return
                    async with slots:
                        results[product_id] = await read_product(product_id)
                finally:
                    queue.task_done()
        finally:
            print("consumer closed")

    async with asyncio.TaskGroup() as group:
        for _ in range(4):
            group.create_task(consume())
        for product_id in product_ids:
            await queue.put(product_id)
        for _ in range(4):
            await queue.put(None)
        await queue.join()
    return results

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。补完 collect_products 后可沿代码数出 4 个消费者、队列容量 4、同时读取上限 2；下一步才执行。

刚才发生了什么

队列存还没处理的编号，消费者取出编号做事，Semaphore 像两张通行证限制正在读取的数量。TaskGroup 把这些消费者当一组管理；finally 的 task_done 让失败项也结清计数。

3. 给整个批次设置截止时间

追加入口，执行 python bounded_async.py。正常预算 3 秒足够处理 20 条；把预算改成 0.15 秒复现取消。TimeoutError 在 timeout 上下文外处理，不能返回一个看似完整的部分结果。

文件：bounded_async.py

async def main():
    try:
        async with asyncio.timeout(3):
            rows = await collect_products(range(20))
        print("completed", len(rows))
    except TimeoutError:
        print("batch timed out")

if __name__ == "__main__":
    asyncio.run(main())

你应该看到什么

正常运行打印 completed 20 和四次 consumer closed；将预算改为 0.15 后打印 batch timed out，消费者仍清理退出。

刚才发生了什么

timeout 给整个批次设截止时间，超时通过取消正在等待的协程来停止；退出上下文后才转成 TimeoutError。只在收集全部成功后输出 completed，避免把半份清单当成完整结果。

4. 复现单条错误与协作取消

临时在 read_product 中加入下面两行，再次执行。TaskGroup 失败应停止整批，其余消费者均输出清理日志。随后恢复函数；如果产品允许部分成功，应另行定义逐条错误结果，不能默默吞掉异常。

文件：bounded_async.py

# 临时插入 read_product 的 await 之前：
# if product_id == 3:
#     raise ValueError("invalid product for drill")

# 如果明确需要观察组合异常，可在入口外围写：
# try:
#     asyncio.run(main())
# except* ValueError as group:
#     print("batch failed", len(group.exceptions))

你应该看到什么

在编号 3 触发 ValueError 后，不出现 completed 20；可见组合异常及各消费者的 consumer closed，恢复后再次正常完成。

刚才发生了什么

一项失败时 TaskGroup 会取消同组其余任务，并把异常汇总交给调用者。协作取消指任务在 await 处响应取消，而不是外部随时强杀一段 Python 代码。

检查学习结果

• 正常分支返回20条对应编号数据，只有固定四个消费者。
• 0.15秒总预算触发 batch timed out，四个消费者执行清理。
• 单条 ValueError 导致整批失败，未返回伪造的完整结果。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

一次导出需要读取大量商品，不能耗尽连接池和内存。

实现决策

• 队列上限、消费者数和I/O并发数分别配置。
• 整批失败与部分成功必须由接口契约决定。
• 总截止时间与每次I/O超时同时存在，超时不代表外部请求未生效。

隔离环境演练

在本地第3条输入抛出异常，再用0.15秒预算运行，记录两种情况下消费者关闭与结果行为。

恢复验收

• 并发峰值≤2、排队容量≤4。
• 异常和超时都没有遗留未等待任务。
• 不会把部分数据标记为完整导出。

遇到问题时查这里

用了Semaphore仍消耗大量内存

原因：先为全部商品创建任务，再限制内部I/O。

处理：改为固定消费者和有界Queue，生产端await put自然等待。

超时后批次仍持续运行

原因：子协程吞掉CancelledError或长期没有await。

处理：清理后重新抛出取消，阻塞I/O移到下一课的合适执行器。

自己试一次

给 read_product 加运行中计数与峰值，验证最大同时读取数不超过2。

提示：计数增加、减少包围真实I/O，减少放finally。

参考答案：

在Semaphore内部增加计数并更新峰值；try/finally中减少。正常与超时路径峰值均≤2，结束后运行中计数为0。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

并发数与任务数量是两种限制

完成第 2 步“建立固定消费者与有限缓冲”后，再回顾这条规则：Semaphore 控制同时进入临界区的协程数量，却不会阻止一次创建十万个 Task。有限 Queue 加固定数量消费者，同时限制等待项和任务对象数量。

TaskGroup 管理一组相关任务

完成第 2 步“建立固定消费者与有限缓冲”后，再回顾这条规则：退出 TaskGroup 时等待子任务；子任务出现非取消异常时会取消其余任务，并以 ExceptionGroup 传播。它不会把任务保存到磁盘，也不负责服务重启后恢复。

超时通过取消实现

完成第 3 步“给整个批次设置截止时间”后，再回顾这条规则：asyncio.timeout() 超时后在上下文外捕获 TimeoutError。协程在 await 等位置响应取消，finally 适合清理；捕获 CancelledError 后应重新抛出。

技术文档

• Python 3.11 TaskGroup 与 timeout
• asyncio.Queue

---

03 区分 BackgroundTasks 与 create_task 的生命周期

上一课能受理导出却始终排队。现在把一个短导出放到响应之后运行，再观察重启服务时记录会怎样。

从一个问题开始

上一课能受理导出却始终排队。现在把一个短导出放到响应之后运行，再观察重启服务时记录会怎样。

这次的终点

短导出能走到 succeeded；独立 Task 实验能取消并清理；重启后旧任务查询 404，识别本地后台工作的边界。

开始前，确认你在这里

• 已完成任务第 01–02 课，tasks_lab 中保留 api_states.py；先停止占用 8000 端口的旧练习服务，再从同一虚拟环境启动本课入口。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 在响应之后执行短导出，并轮询到 succeeded
• 显式登记 create_task 创建的工作，取消后核对清理与剩余数量
• 重启本机服务后查询旧任务，说明内存后台工作的恢复边界

一步步完成案例

1. 运行响应后的小任务

新建 api_background.py，复用第01课的app和状态。这里新增独立路径，避免覆盖原来只受理的/exports。用 python -m uvicorn api_background:app 启动；此实验不要开reload以免编辑时无意重启。

文件：api_background.py

import asyncio
from uuid import uuid4
from fastapi import BackgroundTasks, Response
from api_states import app, jobs, transition, ExportIn

async def local_export(job_id: str):
    transition(job_id, "running")
    try:
        await asyncio.sleep(1)
        jobs[job_id]["result"] = {"rows": len(jobs[job_id]["product_ids"])}
        transition(job_id, "succeeded")
    except asyncio.CancelledError:
        transition(job_id, "cancelled")
        raise
    except Exception as exc:
        jobs[job_id]["error"] = type(exc).__name__
        transition(job_id, "failed")

@app.post("/exports/background", status_code=202)
async def submit_local(body: ExportIn, background: BackgroundTasks, response: Response):
    job_id = str(uuid4())
    jobs[job_id] = {"id": job_id, "status": "queued",
                    "product_ids": body.product_ids, "result": None, "error": None}
    background.add_task(local_export, job_id)
    response.headers["Location"] = f"/exports/{job_id}"
    return jobs[job_id]

你应该看到什么

服务启动后提交 /exports/background 得到 202；短暂轮询 /exports/{id} 可见 running，约 1 秒后为 succeeded，result.rows 为 2。

刚才发生了什么

BackgroundTasks 在响应发出后执行登记的函数，用户不必等函数结束才收到编号。工作仍在当前应用进程里，sleep 只是模拟生成两行清单。

2. 追踪自行创建的Task

新建 task_registry.py。done回调既移除强引用，也获取异常；不要在回调里再次抛出未处理异常。这个组件尚未暴露提交路由，下一步只在独立生命周期实验里使用。

文件：task_registry.py

import asyncio
import logging

running: set[asyncio.Task] = set()

def observe(task):
    running.discard(task)
    if task.cancelled():
        return
    error = task.exception()
    if error is not None:
        logging.error("local task failed: %r", error)

def start(coro):
    task = asyncio.create_task(coro)
    running.add(task)
    task.add_done_callback(observe)
    return task

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。task_registry.py 定义 running 集合和 start；导入它不会自行启动任务，后续完成的 Task 会被 observe 移出集合。

刚才发生了什么

create_task 立即安排协程运行，却不替我们登记退出责任。保存引用便于停机时找到它，读取 exception 则让失败不至于成为无人处理的后台异常。

3. 关闭时等待，再协作取消

新建 registry_demo.py独立运行。先给3秒自然完成窗口，剩余任务发取消并等待清理。此例子仅含遵守取消的sleep；不能对任意阻塞库宣称3秒后必定退出。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python registry_demo.py，再与下方预期对照。

文件：registry_demo.py

import asyncio
from task_registry import start, running

async def work():
    try:
        await asyncio.sleep(30)
    finally:
        print("cleanup")

async def main():
    start(work())
    await asyncio.sleep(0.1)
    pending = set(running)
    if pending:
        _, pending = await asyncio.wait(pending, timeout=3)
    for task in pending:
        task.cancel()
    await asyncio.gather(*pending, return_exceptions=True)
    print("remaining", len(running))

if __name__ == "__main__":
    asyncio.run(main())

你应该看到什么

python registry_demo.py 等待约 3 秒后打印 cleanup，最终 remaining 0，脚本返回终端。

刚才发生了什么

先等自然完成、再发取消、最后 gather 等清理，分别处理三件事。cancel 是停止请求；看到 finally 的 cleanup 和集合归零，才证明本例的协程已退出。

4. 记录正常停止和崩溃的差异

提交 /exports/background 并轮询第01课查询路径，观察running到succeeded。再将sleep临时改为30秒，仅在本地终止该实验进程并重启；原job_id查询404。比较正常关闭可能等待与强制退出丢失，随后恢复1秒。

文件：终端 / 当前课时

POST /exports/background
Content-Type: application/json

{"product_ids":[1,2]}

# 保存202响应中的id，随后查询：
GET /exports/{job_id}

你应该看到什么

正常提交最终能查到 succeeded；仅终止并重启本课实验服务后，同一旧 id 返回 404。

刚才发生了什么

内存字典随进程退出而消失。正常关闭可能给短任务留下完成时间，强制退出没有这个保证；需要跨重启恢复的导出，要把状态和执行移出这一进程。

检查学习结果

• 短任务在响应后运行，查询最终为succeeded。
• registry_demo输出cleanup，且未保留运行中的Task。
• 重启使内存任务消失，实验记录不把它解释为可自动恢复。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

接口响应后写一条可丢失的诊断信息，与必须完成的用户导出同时存在。

实现决策

• 短且允许丢失的工作可留在应用进程。
• 用户导出不能只依赖BackgroundTasks或create_task。
• 退出先关闭入口，再给任务有界清理时间，最终保证由持久系统承担。

隔离环境演练

本地延长任务后终止应用，重启查询旧id，验证不存在自动恢复机制。

恢复验收

• Task引用和异常都被回收。
• 失败状态与HTTP受理状态分开呈现。
• 没有将应用进程任务描述为持久队列。

遇到问题时查这里

返回202后异常只出现在终端

原因：响应已发送，后台工作不再控制这次HTTP状态。

处理：将异常转为任务failed并记录关联job_id。

Task exception was never retrieved

原因：create_task任务失败但无人await或读取异常。

处理：保存引用并获取完成异常，或对相关工作使用TaskGroup。

自己试一次

让work在0.1秒后抛出ValueError，确认回调记录异常且running恢复为空。

提示：不要用空except隐藏错误。

参考答案：

将sleep后加raise ValueError('drill')；observe记录错误，回调移除任务引用，程序结束时无未获取异常提示。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

BackgroundTasks 仍属于当前应用进程

完成第 1 步“运行响应后的小任务”后，再回顾这条规则：任务在响应发出后运行，不是持久队列。执行异常无法把已发送的202改成500；重要结果必须通过任务状态和日志记录。

create_task 需要显式管理

完成第 2 步“追踪自行创建的Task”后，再回顾这条规则：保存强引用，读取完成异常，退出时决定等待或取消。只调用create_task不会自动得到重试、限流或跨进程恢复。

进程生命周期不是业务完成保证

完成第 4 步“记录正常停止和崩溃的差异”后，再回顾这条规则：正常关闭可以给任务清理时间；强制终止、崩溃和机器断电不会执行finally。会影响用户资产的任务应进入后续持久队列方案。

技术文档

• FastAPI BackgroundTasks
• Python create_task与取消

---

04 用 Lock 与 Queue 管理线程共享状态

两个线程一起生成商品清单片段，其中一条数据故意失败。我们希望最终总数仍对得上，脚本也能正常结束。

从一个问题开始

两个线程一起生成商品清单片段，其中一条数据故意失败。我们希望最终总数仍对得上，脚本也能正常结束。

这次的终点

20 个编号得到 19 次成功和 1 次失败，两个线程全部退出；改变失败数量后总计仍为 20。

开始前，确认你在这里

• 已完成任务第 01–03 课；在 tasks_lab 虚拟环境终端新建 threaded_queue.py，本课独立运行，不启动 API、Redis 或 MySQL。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 让两个线程通过容量四的队列处理 20 个商品编号
• 在短锁区修改成功与失败计数，确保失败项也结清
• 核对成功数加失败数为 20，错误数量一致且所有线程退出

一步步完成案例

1. 定义共享计数和有限队列

创建 threaded_queue.py。每个输入只是一条商品编号，结果计数属于这个脚本；不要把该锁当成MySQL记录锁。

文件：threaded_queue.py

from queue import Queue
from threading import Lock, Thread
import time

pending = Queue(maxsize=4)
mutex = Lock()
counts = {"done": 0, "failed": 0}
errors = []

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。文件中 pending 容量为 4，counts 的 done/failed 都是 0；尚未创建运行中的线程。

刚才发生了什么

Queue 负责在线程之间安全交接工作；Lock 负责把一次计数修改围起来。它们只保护本进程里的数据，不会锁住商品数据库记录。

2. 实现保证结清的消费者

追加consumer。异常时记录失败，finally中task_done；即使收到结束哨兵也要结清那一次get。模拟耗时发生在锁外，锁中只修改计数。

文件：threaded_queue.py

def consume():
    while True:
        product_id = pending.get()
        try:
            if product_id is None:
                return
            time.sleep(0.05)
            if product_id == 7:
                raise ValueError("bad product for drill")
            with mutex:
                counts["done"] += 1
        except Exception as exc:
            with mutex:
                counts["failed"] += 1
                errors.append((product_id, type(exc).__name__))
        finally:
            pending.task_done()

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。消费者包含编号 7 的失败分支，成功和失败均经过 task_done；等待时间位于 with mutex 外面。

刚才发生了什么

with mutex 让同时来的两个线程轮流修改计数。finally 保证每次取出的任务都结清，即使失败或读到结束标记，也不会让等待完成的一方一直卡住。

3. 启动、入队与退出

追加并执行脚本。先发送全部编号，再发送两个哨兵；队列满时put等待消费者释放容量。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python threaded_queue.py，再与下方预期对照。

文件：threaded_queue.py

if __name__ == "__main__":
    threads = [Thread(target=consume) for _ in range(2)]
    for thread in threads:
        thread.start()
    for product_id in range(20):
        pending.put(product_id)
    for _ in threads:
        pending.put(None)
    pending.join()
    for thread in threads:
        thread.join()
    print(counts, errors)

你应该看到什么

python threaded_queue.py 打印 {'done': 19, 'failed': 1} 和编号 7 的 ValueError，随后返回提示符。

刚才发生了什么

put 在队列满时等待，因此待处理列表不会无限增大；每个线程收到一个 None 才退出。队列 join 等任务计数清零，线程 join 再等执行线程本身结束。 非daemon线程让退出路径可见。

4. 验证失败项不会卡住队列

预期done=19、failed=1、错误编号7。再把失败条件改成两个编号，确认总数仍20。不要用empty()判断所有任务完成：队列已空时消费者仍可能在处理。

文件：threaded_queue.py

# 验收断言可放在所有thread.join之后：
assert counts["done"] + counts["failed"] == 20
assert all(not thread.is_alive() for thread in threads)
assert len(errors) == counts["failed"]

你应该看到什么

追加断言后正常退出；把失败条件改为两件商品时，成功数加失败数仍是 20，errors 数量与 failed 相等。

刚才发生了什么

队列暂时为空只说明没有待领取项，不能说明已经取出的工作做完。检查总数、错误条数和线程存活状态，才同时覆盖结果与退出路径。

检查学习结果

• 完成数与失败数合计20，错误记录对应失败编号。
• pending.join能够返回，全部线程退出。
• 模拟耗时发生在mutex保护范围之外。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

同步商品读取库只能在线程中调用，希望稳定处理小批量导出。

实现决策

• 共享状态采用短临界区；每个线程自行管理不支持共享的外部连接。
• 队列明确容量，过载时上游等待或拒绝。
• 线程失败必须成为记录中的失败项，不能静默丢失。

隔离环境演练

让第7条输入抛异常并把Queue容量缩小到1，验证错误收集和任务结清。

恢复验收

• 所有入队项都有成功、失败或明确取消结论。
• join不依赖empty/qsize的瞬间值。
• 线程锁没有被当成跨进程幂等方案。

遇到问题时查这里

queue.join永远不返回

原因：异常或哨兵分支漏了task_done。

处理：让每次成功get都在finally中task_done一次。

两个线程耗时与一个相同

原因：持有锁时执行了sleep或阻塞I/O。

处理：只把共享计数更新放在锁里，I/O放在锁外。

自己试一次

把队列容量改成1，并让消费者每条等待0.2秒，观察生产者入队会等待。

提示：在put前后记录perf_counter，不能只记录消费者耗时。

参考答案：

生产者前几次put后会因队列满等待；消费者释放位置后继续。最终仍完成20项且线程正常退出。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

GIL 不提供业务事务

完成第 1 步“定义共享计数和有限队列”后，再回顾这条规则：字典读改写涉及多个步骤，不能用GIL代替业务锁。锁只保护同一进程中的共享状态，不能协调多个Celery进程或多台机器。

Queue 同步与业务状态分开

完成第 3 步“启动、入队与退出”后，再回顾这条规则：queue.Queue提供线程安全的put/get，有界容量用于限制待处理项。取出后必须对应task_done，join等待的是计数清零，不是线程自然退出。

锁内只做短操作

完成第 2 步“实现保证结清的消费者”后，再回顾这条规则：耗时读取、网络等待、文件写入放到锁外。不要在持锁期间等待另一个需要相同锁的任务，避免死锁和无意义串行化。

技术文档

• Python threading.Lock
• Python Queue

---

05 在线程池中调用阻塞 I/O

已有一个只能同步调用的导出读取函数。用户等超时了，后台函数是否真的停止？先用短等待把这个问题看清。

从一个问题开始

已有一个只能同步调用的导出读取函数。用户等超时了，后台函数是否真的停止？先用短等待把这个问题看清。

这次的终点

看到等待超时后线程仍需响应停止标记；批处理最多保留四个 Future，异步调用侧也能提出协作停止。

开始前，确认你在这里

• 已完成任务第 04 课，理解线程与共享停止标记；在 tasks_lab 虚拟环境终端运行 pool_io.py 等独立脚本，无需启动其他服务。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 观察等待 Future 超时后，运行中的线程并未自动停止
• 通过 Event 请求工作函数在分段边界退出
• 分批限制待执行 Future，并在 to_thread 超时后保留同样的停止机制

一步步完成案例

1. 建立同步I/O模拟函数

创建 pool_io.py。函数每小段等待后检查停止事件；本例0.1秒是可控制的模拟I/O，真实库须使用其自身timeout参数。

文件：pool_io.py

import time
from threading import Event
from concurrent.futures import ThreadPoolExecutor, TimeoutError

def read_sync(product_id, stop):
    for _ in range(20):
        if stop.is_set():
            return {"id": product_id, "cancelled": True}
        time.sleep(0.1)
    return {"id": product_id, "cancelled": False}

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。read_sync 一次最多循环 20 段，每段等待 0.1 秒；stop 已设置时返回 cancelled=True，不再等完整 2 秒。

刚才发生了什么

Event 是线程间可共同读取的停止标记。把长工作分成能检查标记的小段，才能让工作函数自己决定尽快退出。

2. 观察运行中Future不能被取消

追加并执行。只有一个任务提交到一个线程，确保实验没有无限排队。0.2秒时通常已经开始执行，cancel返回False；设置stop后等待实际退出。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python pool_io.py，再与下方预期对照。

文件：pool_io.py

if __name__ == "__main__":
    stop = Event()
    with ThreadPoolExecutor(max_workers=1) as pool:
        future = pool.submit(read_sync, 1, stop)
        try:
            print(future.result(timeout=0.2))
        except TimeoutError:
            print("wait timed out; cancel:", future.cancel())
            stop.set()
            print("worker result:", future.result(timeout=1))
    print("pool closed")

你应该看到什么

通常打印 wait timed out; cancel: False，随后 worker result 中 cancelled 为 True，最后 pool closed；若任务尚未启动则取消可能成功，需按实际状态记录。

刚才发生了什么

Future 表示将来会得到的结果。result 的 timeout 只限制调用者等多久，cancel 只能取消尚未运行的任务；stop.set 才让本例已经运行的函数在下一段检查时结束。

3. 按批限制待执行Future

另建 batch_io.py，导入同一同步函数。每次只提交四项，完成后才提交下一批；这会牺牲少量流水线利用率，但不会无限占用内存。不要把本例的循环直接放进async路由阻塞事件循环。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python batch_io.py，再与下方预期对照。

文件：batch_io.py

from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Event
from pool_io import read_sync

def export_batches(product_ids):
    stop = Event()
    results = []
    with ThreadPoolExecutor(max_workers=2) as pool:
        for start in range(0, len(product_ids), 4):
            batch = product_ids[start:start + 4]
            futures = [pool.submit(read_sync, item, stop) for item in batch]
            for future in as_completed(futures):
                results.append(future.result())
    return results

if __name__ == "__main__":
    print(len(export_batches(list(range(8)))))

你应该看到什么

python batch_io.py 最终打印 8；代码中一次只保留 4 个 futures，线程池同时最多执行 2 项。

刚才发生了什么

线程数限制正在做事的人数，提交批次限制排队的工作量。as_completed 按完成次序交回结果，因此结果列表不承诺与输入顺序一致。

4. 在异步调用侧保留同样的退出边界

在独立脚本中用asyncio.to_thread调用read_sync，并在超时后设置Event。to_thread使用的底层线程不会因为await被取消就自动消失；业务确认停止必须来自工作函数结束或持久任务状态。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python async_thread.py，再与下方预期对照。

文件：async_thread.py

import asyncio
from threading import Event
from pool_io import read_sync

async def demo():
    stop = Event()
    try:
        async with asyncio.timeout(0.2):
            return await asyncio.to_thread(read_sync, 1, stop)
    except TimeoutError:
        stop.set()
        print("stop requested, thread exits cooperatively")

if __name__ == "__main__":
    asyncio.run(demo())

你应该看到什么

独立异步脚本打印 stop requested, thread exits cooperatively 并最终退出；不会在 0.2 秒时直接得到完整读取结果。

刚才发生了什么

to_thread 把同步函数放到线程执行，让事件循环继续工作。取消 await 不会强杀该线程，所以超时分支仍要设置停止标记，退出结论仍取决于工作函数是否遵守它。

检查学习结果

• 等待超时被捕获，运行中的cancel返回False。
• 协作停止后工作结果为cancelled=True，线程池最终关闭。
• 批处理每批最多4个Future，同时执行最多2个。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

导出调用只有同步接口的存储SDK，接口请求可能先超时。

实现决策

• 同步SDK必须有连接和读写超时。
• 线程数、待提交数量与外部服务额度一起限制。
• 超时之后的外部副作用依靠幂等与核对，不假定取消已经回滚。

隔离环境演练

本地让read_sync运行2秒，0.2秒停止等待，记录cancel返回值和线程实际结束时间。

恢复验收

• 没有声称Future.cancel能强杀线程。
• 线程池退出后没有遗留工作线程。
• 超时状态不被误当成外部操作未执行。

遇到问题时查这里

接口已返回超时，外部写入却仍发生

原因：调用方取消了等待，没有停止底层线程。

处理：外部操作需要幂等键和明确状态，给库设置超时并使用协作取消。

离开with仍长时间卡住

原因：线程池正在等待没有超时的阻塞调用完成。

处理：先修正I/O本身的时间上限，不把shutdown当强制终止接口。

自己试一次

增加一个尚未运行的Future，验证排队任务cancel可能为True，而正在运行的任务为False。

提示：max_workers=1，先提交一个长任务，再立即提交第二个。

参考答案：

第二个尚未运行时cancel返回True，不能再对它当作成功结果调用；首个仍需stop.set并等待实际退出。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

线程池只限制同时执行数

完成第 3 步“按批限制待执行Future”后，再回顾这条规则：max_workers不会限制任意多次submit累积的待执行Future。调用方仍需限批提交、有界队列或接入限流。

Future超时不等于函数停止

完成第 2 步“观察运行中Future不能被取消”后，再回顾这条规则：future.result(timeout=...)只限制调用方等待时间。cancel只能取消尚未运行的工作，无法安全强杀运行中的线程。

取消需要协作与I/O超时

完成第 4 步“在异步调用侧保留同样的退出边界”后，再回顾这条规则：给同步库设置超时，长循环定期检查Event。with退出线程池默认等待工作完成，不能靠离开with逃避仍在运行的阻塞函数。

技术文档

• ThreadPoolExecutor与Future
• asyncio.to_thread

---

06 用进程池处理 CPU 密集导出

清单导出需要大量文字计算，没有网络可等待。本课比较直接计算与交给两个独立进程，先检查结果一致再看耗时。

从一个问题开始

清单导出需要大量文字计算，没有网络可等待。本课比较直接计算与交给两个独立进程，先检查结果一致再看耗时。

这次的终点

串行与进程池给出相同校验结果，并打印各自耗时；能识别不能跨进程传递的函数和工作进程失败。

开始前，确认你在这里

• 已完成任务第 05 课；在 tasks_lab 的终端运行实际 .py 文件，先确认可导入同目录文件；不要在交互式 Console 中定义工作函数。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 把文字计算函数放到可由子进程导入的模块中
• 比较串行与两个进程的实测耗时，并断言计算结果相同
• 辨认局部函数序列化错误和工作进程失效时的整批失败

一步步完成案例

1. 写一个可导入的CPU函数

新建 cpu_work.py。使用纯Python循环产生可比较的校验结果，不借助网络或数据库。规模可以调小，不能为了展示加速耗尽机器资源。

文件：cpu_work.py

def score_text(text: str, rounds: int) -> int:
    total = 0
    for _ in range(rounds):
        for character in text:
            total = (total * 33 + ord(character)) % 1000000007
    return total

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。cpu_work.py 只定义 score_text；对相同文本和 rounds 调用时返回相同整数，不访问数据库或网络。

刚才发生了什么

这里的耗时来自 Python 循环计算，而不是等待 I/O。把函数放到模块顶层，新的进程才可以通过导入模块找到它。

2. 比较串行与进程池

创建 process_export.py。在终端运行文件，避免直接在交互式Console定义worker函数。先用两个进程、小输入测量；小任务可能更慢，这是有效结果。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python process_export.py，再与下方预期对照。

文件：process_export.py

import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor
from time import perf_counter
from cpu_work import score_text

def main():
    texts = ["product export " * 100 for _ in range(4)]
    started = perf_counter()
    serial = [score_text(text, 2000) for text in texts]
    print("serial", perf_counter() - started)
    started = perf_counter()
    with ProcessPoolExecutor(max_workers=2, mp_context=mp.get_context("spawn")) as pool:
        parallel = list(pool.map(score_text, texts, [2000] * len(texts)))
    print("process", perf_counter() - started)
    assert serial == parallel

if __name__ == "__main__":
    main()

你应该看到什么

python process_export.py 打印 serial 与 process 两个耗时，assert 不报错；进程版可能更慢，应保留实际测量。

刚才发生了什么

进程池让不同解释器分担计算，spawn 从新解释器启动。传参和启动也要花时间，小任务的额外成本可能超过并行收益；入口保护防止子进程导入时再次建池。

3. 复现不可序列化与启动错误

把目标函数临时改为main内部的局部函数，观察序列化失败，然后恢复模块顶层定义。不要删除入口保护后在繁忙机器试验递归启动；用代码审查确认创建池只发生在受保护入口。

文件：process_pool_review.py

# 错误结构，只用于辨认，勿替换为实际入口：
# def main():
#     def local_score(text):
#         return len(text)
#     pool.submit(local_score, "product")

# 正确结构：
from cpu_work import score_text
assert callable(score_text)

你应该看到什么

局部函数提交实验出现 pickle/序列化相关错误；恢复为模块顶层 score_text 后第二步重新通过。

刚才发生了什么

进程间不能直接共享函数所在的内存，需要把参数和函数定位信息转换成可传输形式，这叫序列化。函数可调用不等于它能从另一进程重新找到。

4. 给进程失败定义批次结论

在独立调用侧捕获BrokenProcessPool并把整批视为失败。Future等待超时也不会自动结束已运行CPU函数；大工作应分块，控制单块长度，避免在API每次请求里创建一整池进程。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python verify_process.py，再与下方预期对照。

文件：verify_process.py

from concurrent.futures.process import BrokenProcessPool
from process_export import main

if __name__ == "__main__":
    try:
        main()
    except BrokenProcessPool:
        print("batch failed: worker process lost")
        raise

你应该看到什么

正常入口仍打印两组耗时；工作进程意外失效时捕获 BrokenProcessPool 并报告 batch failed，再把错误交回调用者。

刚才发生了什么

进程池损坏意味着这批结果不能再假定完整。等待超时也不代表 CPU 函数被终止，因此应控制单块计算长度，并在整批层面明确失败结果。

检查学习结果

• serial与parallel结果一致。
• 脚本使用显式spawn、顶层目标函数和主入口保护。
• 记录实际耗时，不把进程池预设为所有输入都更快。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

商品导出需要大量纯Python清洗或统计，阻塞API事件循环。

实现决策

• 先测量CPU占比，再选择进程数。
• 传入有限数据块，不共享数据库连接。
• 池在受控生命周期中管理，失败批次进入可恢复任务流程。

隔离环境演练

仅在隔离脚本把目标函数换为局部定义，确认序列化失败可见且不会产出成功结果。

恢复验收

• 结果可与串行版本交叉验证。
• CPU工作没有被描述成async能够自动加速。
• 跨进程输入和启动方式清晰可复现。

遇到问题时查这里

Can't pickle local object

原因：任务函数是闭包、lambda或局部定义。

处理：把函数移入可导入模块顶层，仅传可序列化参数。

每个API请求创建进程池导致CPU和内存暴涨

原因：并发请求各自放大worker数量。

处理：使用受控的应用级池或后续独立Celery进程，限制整体并发。

自己试一次

把rounds分别设为10和2000，比较小任务与大任务的进程启动占比。

提示：两次保持输入、进程数与计时范围一致。

参考答案：

两个版本均先验证结果相同，再记录耗时。小任务可能明显更慢；只有在任务足够大时，多核收益才可能覆盖序列化与启动成本。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

CPU工作与I/O等待不同

完成第 1 步“写一个可导入的CPU函数”后，再回顾这条规则：在通常带GIL的CPython构建中，多个线程不会让纯PythonCPU循环稳定获得多核并行；某些C扩展会释放GIL，需实际测量。进程池提供独立解释器，代价是启动与数据复制。

参数需要跨进程序列化

完成第 3 步“复现不可序列化与启动错误”后，再回顾这条规则：传可序列化的文本、编号和结构，不传Request、锁、数据库连接或打开的文件。每个进程需要自己的连接生命周期。

显式spawn避免依赖平台默认值

完成第 2 步“比较串行与进程池”后，再回顾这条规则：Python不同版本与平台的默认启动方式不同。本模块明确使用spawn，目标函数放模块顶层，并用if name == 'main'保护启动代码。

技术文档

• ProcessPoolExecutor
• multiprocessing与spawn

---

07 选择协程、线程、进程或独立任务队列

同样是导出清单，有的在等网络，有的在算文字，还有的必须在关闭网页后继续。现在用前六课的现象决定工作放在哪里执行。

从一个问题开始

同样是导出清单，有的在等网络，有的在算文字，还有的必须在关闭网页后继续。现在用前六课的现象决定工作放在哪里执行。

这次的终点

得到一份按实际工作约束选择执行方式的 decision.md，以及可传递、大小有上限的商品快照消息。

开始前，确认你在这里

• 已完成任务第 01–06 课并保留 bounded_async.py、pool_io.py、process_export.py；在 tasks_lab 虚拟环境终端核对这些文件存在。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 根据网络等待、同步调用、CPU 计算和跨重启要求选择执行位置
• 把商品快照编码成小于 64KB、无连接对象或凭据的 JSON 消息
• 记录 Celery 提交者、中转处、执行者与持久状态表的职责

一步步完成案例

1. 把四类工作写成可检验的需求

建立decision.md，逐项填写现有shop-service的工作量，而不是只写“异步”。本例行数是练习限制，不是普遍容量建议。

文件：decision.md

| 工作 | 主要约束 | 起点 |
|---|---|---|
| 同一请求并发读20条商品 | 可等待I/O、请求内完成 | asyncio + 有界队列 |
| 同步SDK上传一个导出文件 | 阻塞I/O、有超时 | 受限线程池 |
| 大批纯Python文本统计 | CPU、数据可分块 | 受限进程池 |
| 关闭API后仍需完成导出 | 持久状态、独立消费者 | 任务队列 + 状态表 |

你应该看到什么

decision.md 有四行具体工作、约束和对应起点，能指出哪一项要求 API 退出后仍继续。

刚才发生了什么

协程、线程和进程首先解决工作在哪里执行；跨重启恢复还要求独立执行者和持久记录，单纯增大并发数解决不了记录消失。

2. 固定消息边界

在当前 tasks_lab 目录保存 message_contract.py，运行 python message_contract.py。只传 job_id 与本次固定的小型商品快照，不把密码、请求对象或数据库连接放入消息。保留所有五项商品字段和整数分 price_cents，后续重投使用同一快照。

文件：message_contract.py

import json
from uuid import uuid4

message = {
    "job_id": str(uuid4()),
    "rows": [{"id": 1, "name": "product", "description": "", "is_active": False, "price_cents": 0}],
}
encoded = json.dumps(message, ensure_ascii=False).encode("utf-8")
assert len(encoded) < 64 * 1024
print(len(encoded))

你应该看到什么

保存 message_contract.py 并运行后打印一个小于 65536 的字节数，JSON 中只有 job_id 和简单商品字段。

刚才发生了什么

JSON 把内容变成别的进程也能解析的文本。小型快照固定本次导出的输入；连接、请求对象和凭据既不是业务输入，也不应被塞进任务消息。

3. 比较三种队列的实际运行边界

阅读下方官方资料，记录所选版本的broker、worker启动方式、超时、重试和结果保留方式。只安装下一课Celery，不同时引入三套消费者。Celery5.x不把原生Windows作为支持平台；本模块worker统一在Linux或WSL2运行。

文件：decision.md

# 记录到decision.md：
# broker: Redis 7，仅本机隔离练习
# worker: Celery 5.x / Linux 或 WSL2
# durable status: MySQL 8.4 export_jobs
# delivery: 允许重复，业务处理必须幂等
# result backend: 临时查询缓存，不能替代export_jobs

你应该看到什么

decision.md 记录 Redis broker、Celery worker 的 Linux/WSL2 运行方式及 MySQL export_jobs 持久状态位置。

刚才发生了什么

broker 是暂存待交付消息的中转处，worker 是独立取消息执行的进程，结果缓存只短期保存执行回报。三者分开后，API 才能退出而不带走整个执行过程。

4. 把迁移条件写进验收

运行前几课的小脚本并记录：进程退出是否丢任务、CPU是否饱和、待处理数量是否有界。若要求跨重启恢复，不以继续增加线程数作为解决方案；从下一课分离producer与worker。

文件：终端 / 当前课时

python bounded_async.py
python pool_io.py
python process_export.py
python message_contract.py

你应该看到什么

四个脚本均有实际运行记录，decision.md 写出是否丢任务、计算耗时和排队上限；未做的测试明确记为未验证。

刚才发生了什么

执行方式应由观察支持：需要持久恢复时迁到独立 worker，需要限制等待量时设置有界队列。下一课只搭建一条 Celery 链路，以便验证这些职责。

检查学习结果

• 四类工作都有明确执行位置和适用边界。
• 消息能够JSON编码且没有连接对象或凭据。
• Celery运行平台与状态真源已写入decision.md。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

从单机小导出演进到API与导出worker独立运行。

实现决策

• 工具选择同时考虑运行平台、维护成本和失败恢复。
• 统一JSON消息和业务job_id，降低执行器替换成本。
• 队列不是状态数据库，外部输出还需幂等处理。

隔离环境演练

对比结束本地线程脚本与停止独立worker后的任务存留，用观察结果修订选型表。

恢复验收

• 没有用异步语法替代持久交付设计。
• 平台约束明确，worker不按原生Windows支持说明。
• 迁移的需求来自可测量约束。

遇到问题时查这里

用了Celery就宣称任务只执行一次

原因：忽略ack、重投和外部副作用窗口。

处理：按允许重复设计，后续用幂等输入与状态CAS约束结果。

把Result TTL当任务生命周期

原因：缓存过期被误读为任务不存在。

处理：查询任务真源使用MySQL，结果缓存只用于有限期辅助诊断。

自己试一次

给“同步SDK耗时3秒且必须跨API重启完成”选择方案。

提示：执行方式和交付方式可组合，并非只能选一个。

参考答案：

独立队列承诺持久受理与恢复；worker内部按SDK性质使用同步调用或受限线程。仍须设置SDK超时、幂等键和持久状态。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

执行器回答工作在哪里运行

完成第 1 步“把四类工作写成可检验的需求”后，再回顾这条规则：asyncio适合可等待I/O，线程承接阻塞I/O，进程承接需要隔离或多核的工作。它们本身不提供机器重启后的任务恢复。

任务队列增加独立消费者

完成第 3 步“比较三种队列的实际运行边界”后，再回顾这条规则：RQ围绕Redis/Valkey队列和Python函数，Dramatiq围绕actor与中间件，Celery提供任务路由、重试、调度等能力。不同产品都需要明确版本、broker与故障交付语义。

本模块选择Celery并保留业务状态

完成第 3 步“比较三种队列的实际运行边界”后，再回顾这条规则：主线采用Celery5.x、Redis7作为练习broker与结果缓存，MySQL 8.4 LTS保存任务真源。选择工具不能消除重复执行、外部副作用和数据库一致性问题。

技术文档

• RQ官方文档
• Dramatiq官方指南
• Celery First Steps
• Celery平台FAQ

---

08 启动 Celery broker、worker 与结果缓存

要让商品清单在提交终端退出后继续生成，先把提交者、消息中转和执行者分开，跑通一条只有一件商品的链路。

从一个问题开始

要让商品清单在提交终端退出后继续生成，先把提交者、消息中转和执行者分开，跑通一条只有一件商品的链路。

这次的终点

一个终端提交快照，另一个 worker 生成 artifacts 中的 JSON 文件，提交端查到结果和 SUCCESS。

开始前，确认你在这里

• 已完成任务第 07 课；进入 Linux/WSL2 的 shop-service/tasks_lab 并激活其虚拟环境；先用 docker version 确认 Docker 可访问，本课使用独立端口 6384。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。
• worker、producer、publisher 均在同一个 Linux/WSL2 环境的 tasks_lab 启动；Bash 用 source .venv/bin/activate。环境变量只在当前终端生效，新终端按本课文件加载 .env.tasks，不复制 Windows 虚拟环境。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 在 Linux/WSL2 启动独立认证 Redis，并用 PING 验证连接
• 启动能注册 products.export 的 Celery worker
• 从另一个终端提交快照，核对 SUCCESS、输出路径与真实文件内容

一步步完成案例

1. 准备 Linux/WSL2 环境与独立 Redis

Windows 学习者使用 WSL2；若前 7 课在 Windows 完成，先把 tasks_lab 的 .py 文件复制到 WSL2 Linux 文件系统的 shop-service/tasks_lab。在该目录执行 python3 -m venv .venv（已有则复用）和 source .venv/bin/activate，不复制 Windows 的 .venv。用 python -c "import sys; print(sys.executable)" 核对解释器，再执行下面命令。docker version 必须有服务端信息；Windows 可启用 Docker Desktop 的 WSL 集成。核对专用 6384 端口与 shop-task-broker 名称未被其他项目占用，不删除未知容器。

文件：终端 / 当前课时

python -m pip install "celery[redis]>=5.5,<6" "PyMySQL[rsa]>=1.1,<2" "python-dotenv>=1,<2"
docker version
docker ps --format 'table {{.Names}}\t{{.Ports}}'
python -m pip freeze > requirements-tasks-lock.txt

你应该看到什么

依赖安装完成，docker version 有服务端信息，docker ps 可核对 6384 与 shop-task-broker 是否已占用。

刚才发生了什么

Celery 是独立执行任务的框架；redis 依赖负责和中转实例通信。先确认环境与端口，避免把消息误投到商品缓存练习的 6380。

2. 创建仅用于本地练习的认证配置

保存 prepare_broker.py，下一步运行。它生成 TASK_BROKER_PASSWORD、带认证的 Celery URL 和 Redis 配置，不打印凭据，也不会覆盖现有文件。broker-local 目录只允许当前宿主用户进入；配置文件本身可由容器内 Redis 用户读取。请在 WSL 的 Linux 文件系统中运行，避免 Windows 挂载盘权限语义差异。把 .env.tasks 与 broker-local/ 加入当前项目 .gitignore；不要提交、分享配置或把 URL 复制进日志。两个 Redis DB 只分离键空间，仍共享容量与可用性。

文件：prepare_broker.py

from pathlib import Path
import secrets
from urllib.parse import quote

folder = Path("broker-local")
env_file = Path(".env.tasks")
if env_file.exists() or folder.exists():
    raise SystemExit("local broker configuration exists; reuse it")
folder.mkdir(mode=0o700)
password = secrets.token_urlsafe(24)
encoded = quote(password, safe="")
env_file.write_text(
    f"TASK_BROKER_PASSWORD={password}\n"
    f"CELERY_BROKER_URL=redis://:{encoded}@127.0.0.1:6384/0\n"
    f"CELERY_RESULT_BACKEND=redis://:{encoded}@127.0.0.1:6384/1\n",
    encoding="utf-8",
)
env_file.chmod(0o600)
conf = folder / "redis.conf"
conf.write_text(
    "bind 0.0.0.0\nprotected-mode yes\nport 6379\n"
    f"requirepass {password}\n"
    "dir /data\nappendonly yes\nappendfsync everysec\n"
    "maxmemory 256mb\nmaxmemory-policy noeviction\n",
    encoding="utf-8",
)
conf.chmod(0o644)
ignore = Path(".gitignore")
existing = ignore.read_text(encoding="utf-8") if ignore.exists() else ""
with ignore.open("a", encoding="utf-8") as stream:
    for rule in (".env.tasks", "broker-local/"):
        if rule not in existing.splitlines():
            stream.write("\n" + rule + "\n")
print("local broker files created; credentials were not printed")

你应该看到什么

运行下一步脚本时创建 .env.tasks 和 broker-local/redis.conf，输出 credentials were not printed；已有文件时明确退出而不覆盖。

刚才发生了什么

客户端和 Redis 要使用同一密码才能连接。环境文件供 Python 读取，配置文件供 Redis 读取；两个 DB 编号只分隔键名，仍是同一 Redis 进程。

3. 启动 broker 并验证认证连接

先运行 python preparebroker.py，再执行以下命令。数据使用独立命名卷；Redis 配置只读挂载，本课不执行 CONFIG REWRITE。启动后用 Python 客户端读取 .env.tasks 并 PING，只输出 PONG。若连接仍指向旧端口，打开未设置旧 CELERY* 变量的新终端；load_dotenv 默认保留已注入环境变量。停止后保留容器可用 docker start shop-task-broker 恢复，不重复 docker run。

文件：终端 / 当前课时

python prepare_broker.py
docker run -d --name shop-task-broker \
  -p 127.0.0.1:6384:6379 \
  --mount type=bind,src="$PWD/broker-local/redis.conf",dst=/usr/local/etc/redis/redis.conf,readonly \
  --mount type=volume,src=shop-task-broker-data,dst=/data \
  redis:7 redis-server /usr/local/etc/redis/redis.conf
python - <<'PY'
import os
from dotenv import load_dotenv
from redis import Redis
load_dotenv(".env.tasks")
client = Redis.from_url(
    os.environ["CELERY_BROKER_URL"],
    socket_connect_timeout=2, socket_timeout=2,
)
try:
    print("PONG" if client.ping() else "unexpected reply")
finally:
    client.close()
PY

你应该看到什么

docker run 返回容器 ID，随后验证脚本只打印 PONG；再次学习已有容器时用 docker start 恢复。

刚才发生了什么

PONG 是一次真正的带认证往返，证明地址、端口和密码能配合工作。命名卷把中转实例的数据放在容器可复用的存储位置。 宿主映射限制为 127.0.0.1:6384。

4. 创建Celery应用

新建 celery_app.py，每次从 tasks_lab 目录运行时先加载本地 .env.tasks。部署环境已注入的变量优先，不会被文件覆盖。结果保留 1 小时；worker 只接受 JSON，消息内不传凭据。

文件：celery_app.py

import os
from celery import Celery
from dotenv import load_dotenv

load_dotenv(".env.tasks")

app = Celery(
    "shop_service",
    broker=os.environ["CELERY_BROKER_URL"],
    backend=os.environ["CELERY_RESULT_BACKEND"],
    include=["export_tasks"],
)
app.conf.update(
    task_serializer="json", result_serializer="json",
    accept_content=["json"], result_expires=3600,
    task_default_queue="exports", task_track_started=True,
    broker_connection_timeout=3,
)

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。celery_app.py 可从当前目录读取配置，include 中有 export_tasks，默认队列为 exports，结果保留 3600 秒。

刚才发生了什么

Celery 应用把连接地址、消息格式和任务模块集中在一起。只接受 JSON 可让消息契约清晰；结果保留期说明查询缓存会到期，不是永久业务记录。 此处尚未启用 late ack，下一课演示交付窗口。

5. 实现快照导出并启动worker

新建export_tasks.py。校验job_id，文件只写入当前练习目录的artifacts。另一个相同环境终端启动 python -m celery -A celery_app:app worker -l INFO -Q exports --concurrency=2。确认启动列表出现products.export。

文件：export_tasks.py

import json
from pathlib import Path
from uuid import UUID
from celery_app import app

@app.task(name="products.export")
def export_products(job_id: str, rows: list[dict]):
    safe_id = str(UUID(job_id))
    folder = Path("artifacts")
    folder.mkdir(exist_ok=True)
    target = folder / f"{safe_id}.json"
    target.write_text(
        json.dumps(rows, ensure_ascii=False, sort_keys=True),
        encoding="utf-8",
    )
    return {"job_id": safe_id, "path": str(target), "rows": len(rows)}

你应该看到什么

worker 启动日志的任务列表出现 products.export，进入 ready 等待；收到任务后将在 artifacts 下写出按 job_id 命名的文件。

刚才发生了什么

装饰器给函数一个可跨进程识别的任务名。worker 导入同一模块才知道该名字对应哪个函数，消息中的编号也先按 UUID 校验再用于路径。

6. 从独立producer提交并查询

创建produce_export.py，在有相同环境变量的另一终端运行。get仅用于本地命令行验收，不放进FastAPI请求或Celery任务中同步等待。SUCCESS表示此任务执行结果已写缓存，文件仍位于worker本地路径。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python produce_export.py，再与下方预期对照。

文件：produce_export.py

from uuid import uuid4
from export_tasks import export_products

job_id = str(uuid4())
rows = [{"id": 1, "name": "first", "description": "sample product description", "is_active": False, "price_cents": 0}]
result = export_products.apply_async(args=[job_id, rows], task_id=job_id)
print("submitted", result.id)
print(result.get(timeout=15))
print("state", result.state)

你应该看到什么

python produce_export.py 打印 submitted 编号、含 path/rows=1 的结果和 state SUCCESS；worker 对应路径的 JSON 含本次商品。

刚才发生了什么

apply_async 负责投递，不是在当前终端直接执行导出。get 让本地验收脚本等结果；SUCCESS 仅说明这次执行已报告成功，文件仍属于 worker 所在机器。

检查学习结果

• 认证客户端 PING 返回 PONG，连接目标为独立 broker 6384；worker 注册列表包含 products.export。
• producer获得job_id，worker成功后返回rows=1和相对文件路径。
• artifacts内JSON与提交快照一致，API线程没有等待result.get。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

API提交导出后立即响应，独立worker完成文件生成。

实现决策

• API不调用AsyncResult.get阻塞请求。
• 生产文件保存到所有消费者可访问的持久存储；本地路径只用于单机实验。
• broker持久性、结果TTL和任务状态分别配置并验收。
• 任务 broker 与缓存实例分开；本地 AOF everysec 与 noeviction 仍不能消除丢失窗口，后续用持久状态和 outbox 恢复。

隔离环境演练

仅停止本机练习worker，提交一条后恢复，观察受理、等待超时与最终完成的区别。

恢复验收

• producer和worker确实通过独立进程通信。
• 文件内容对应输入，任务ID能追踪。
• 没有把Redis结果缓存描述为状态真源。

遇到问题时查这里

Received unregistered task

原因：worker未导入模块或producer任务名不一致。

处理：检查include、当前目录与products.export注册名，更新代码后重启worker。

查询随机ID显示PENDING

原因：结果backend对未知或已过期ID可能没有记录。

处理：PENDING不等于确定在排队；第11课使用状态表区分未知与已受理。

连接拒绝、NOAUTH 或一直连旧端口

原因：宿主与容器网络地址混用、没有加载认证 URL，或旧环境变量覆盖 .env.tasks。

处理：宿主使用 127.0.0.1:6384；同一 Compose 网络内使用 broker:6379。确认当前目录、已注入环境和凭据来源，不把完整 URL 打印到日志。

自己试一次

先停止worker再提交一条，记录producer拿到ID但不能立刻完成，恢复worker后再查询。

提示：使用同一broker与队列，不把get的等待超时当作任务执行失败。

参考答案：

暂停期间消息可留在broker；producer等待可能超时，但恢复worker后任务仍可能成功。随后用相同ID核对结果与文件。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

三个角色分别运行

完成第 6 步“从独立producer提交并查询”后，再回顾这条规则：producer提交消息，broker暂存与转交消息，worker执行函数。result backend保存短期执行结果；它的可用性与保留时间不等于业务记录的完整性。

函数导入路径属于部署契约

完成第 5 步“实现快照导出并启动worker”后，再回顾这条规则：worker必须能导入celery_app和export_tasks。装饰器指定稳定任务名，producer与worker使用相同代码和兼容参数。

先验证最小链路

完成第 6 步“从独立producer提交并查询”后，再回顾这条规则：本课发送一条商品快照并生成本地 JSON 文件，暂不读取 MySQL。第 11 课改为从数据库读取任务快照。命令都在同一 Linux/WSL2 环境运行；宿主连接 broker 6384，Docker Compose 内连接 broker:6379，不能混用两种网络地址。

技术文档

• Celery入门
• Celery Redis broker
• Celery配置
• Redis Docker 官方镜像与配置挂载

---

09 理解 ack、重投与幂等输出

文件已经生成，但执行进程在告诉中转处“完成”前退出，消息可能再来一次。让同一个导出重复执行也得到相同文件。

从一个问题开始

文件已经生成，但执行进程在告诉中转处“完成”前退出，消息可能再来一次。让同一个导出重复执行也得到相同文件。

这次的终点

重复交付得到相同逻辑路径与摘要；在隔离故障演练中能区分文件已写入与消息已确认。

开始前，确认你在这里

• 已完成任务第 08 课，broker 的 PING 成功且 worker 已能生成文件；在 tasks_lab 修改 celery_app.py、export_tasks.py 后重启本机 worker。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。
• worker、producer、publisher 均在同一个 Linux/WSL2 环境的 tasks_lab 启动；Bash 用 source .venv/bin/activate。环境变量只在当前终端生效，新终端按本课文件加载 .env.tasks，不复制 Windows 虚拟环境。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 配置完成后确认，并说明执行结束与消息确认之间的窗口
• 重复投递同一业务编号和快照，核对目标路径与摘要相同
• 在本机模拟写完未确认的退出，观察重投与最终文件状态

一步步完成案例

1. 配置确认与可见性时间

在celery_app.py追加配置并重启练习worker。此处最长任务应小于30秒，设置60秒仅便于本地恢复观察；真实配置需覆盖执行上界且考虑恢复等待。

文件：celery_app.py

app.conf.update(
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,
    broker_transport_options={"visibility_timeout": 60, "socket_timeout": 3, "socket_connect_timeout": 3},
    result_backend_transport_options={"visibility_timeout": 60},
    visibility_timeout=60,
)

你应该看到什么

重启 worker 后配置为完成后确认、预取倍数 1，三处 visibility_timeout 都是 60 秒；本地小任务应在 30 秒内完成。

刚才发生了什么

ack 是执行者告诉中转处可以结束本次交付。延后确认可让未确认消息有恢复机会，也让已做完但未确认的工作可能重复；可见性时间决定暂时领走的消息多久后有机会回来。 Redis的三处visibility_timeout统一。

2. 用临时文件完成原子替换

用下面整体替换第08课export_tasks.py。每次执行使用自己的临时文件，最终名字只由job_id决定；输入仍是已固定的小快照。异常时清理本次临时文件，不删除其他任务的输出。

文件：export_tasks.py

import hashlib
import json
import os
from pathlib import Path
from uuid import UUID, uuid4
from celery_app import app

@app.task(name="products.export")
def export_products(job_id: str, rows: list[dict]):
    safe_id = str(UUID(job_id))
    payload = json.dumps(rows, ensure_ascii=False, sort_keys=True).encode("utf-8")
    folder = Path("artifacts")
    folder.mkdir(exist_ok=True)
    target = folder / f"{safe_id}.json"
    temporary = folder / f"{safe_id}.{uuid4()}.tmp"
    try:
        temporary.write_bytes(payload)
        os.replace(temporary, target)
    finally:
        temporary.unlink(missing_ok=True)
    return {"job_id": safe_id, "path": str(target),
            "sha256": hashlib.sha256(payload).hexdigest()}

你应该看到什么

替换后每次写入独有的 .tmp 文件，再替换固定 job_id.json；返回中有 sha256，成功后无本次临时文件残留。

刚才发生了什么

先写临时文件再 os.replace，避免读者看到只写一半的目标文件。同一个编号配合同一份输入生成相同内容，这才是本例重复执行时的结果约束。

3. 重复投递同一业务任务

在独立producer提交两次相同job_id和快照，但使用不同Celery消息ID，避免把共享结果缓存覆盖当作去重。应比较逻辑输出路径与摘要；Celery task_id相同本身也不是业务去重保证。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python duplicate_export.py，再与下方预期对照。

文件：duplicate_export.py

from uuid import uuid4
from export_tasks import export_products

job_id = str(uuid4())
rows = [{"id": 1, "name": "fixed snapshot", "description": "", "is_active": False, "price_cents": 0}]
first = export_products.apply_async(args=[job_id, rows], task_id=str(uuid4()))
second = export_products.apply_async(args=[job_id, rows], task_id=str(uuid4()))
a, b = first.get(timeout=15), second.get(timeout=15)
assert a["path"] == b["path"]
assert a["sha256"] == b["sha256"]
print("same logical output", a)

你应该看到什么

两条不同消息 ID 的任务结果 path 和 sha256 相等，打印 same logical output；artifacts 中逻辑文件仍按同一 job_id 命名。

刚才发生了什么

消息编号标识一次交付，job_id 标识一次业务导出。相同 task_id 本身不会阻止函数再运行，验收必须比较业务输出，而非只比较缓存里的消息状态。

4. 观察重投窗口并记录限制

在本地任务里临时于os.replace后增加5秒sleep，记录执行worker子进程PID，只终止该隔离进程以模拟完成后未确认。重投时再次进入任务；观察日志、最终文件和摘要，恢复后删除故障代码。不要给永久崩溃消息无限恢复机会，后续用持久attempts和死信限制。

文件：终端 / 当前课时

python -m celery -A celery_app:app inspect active
python -m celery -A celery_app:app inspect reserved
# 只对已确认属于本次隔离任务的worker子进程做故障注入。
# 使用日志与inspect中的任务ID核对，不按模糊进程名批量终止。

你应该看到什么

只终止已核对的故障子进程后，日志可见再次执行；最终文件仍可解析且摘要相同，恢复后移除 sleep。

刚才发生了什么

重投说明系统允许重复而不是保证恰好一次。恢复依赖 broker 保留、worker 可用和时间配置；永久崩溃需要后续持久次数上限，不能无限消耗执行资源。

检查学习结果

• 两个不同Celery消息ID携带同一job_id，结果路径和摘要一致。
• 正常读取最终文件得到完整JSON，没有半文件。
• 能够解释late ack的重复窗口和worker丢失时的配置差异。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

导出文件已生成，worker在发送确认前崩溃。

实现决策

• 按可能重复交付设计，每个副作用明确业务键。
• late ack和worker丢失重投策略需与毒消息隔离一起启用。
• 本地原子重命名只解决部分写入观察，数据库中的版本权属在第11课建立。

隔离环境演练

重复消息确定性验收后，再在隔离worker完成输出到ack之间终止执行进程，观察重复进入而非假定只执行一次。

恢复验收

• 重复执行不会产生两个不同逻辑导出资源。
• 输入快照不可变约束被明确记录。
• 失败恢复说明没有宣称绝对不丢失或exactly once。

遇到问题时查这里

固定task_id后仍看到函数执行两次

原因：消息ID或结果缓存键不提供业务级只执行一次保证。

处理：用不可变输入、稳定业务键和后续数据库CAS保护最终结果。

任务持续崩溃并不断重投

原因：reject_on_worker_lost重排同一毒消息，没有持久尝试预算。

处理：限制持久attempts，把不可恢复输入转失败或死信，修复后受控重放。

自己试一次

给第二次投递同一job_id却不同title，观察原子替换无法保证业务版本正确。

提示：本实验故意违反输入不可变约束，只使用新建隔离job_id。

参考答案：

最终文件取决于最后完成者，不能靠os.replace解决。第11课由状态表固定快照，消息仅传job_id，并用租约token防旧执行者覆盖结果引用。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

确认时机会改变丢失与重复窗口

完成第 1 步“配置确认与可见性时间”后，再回顾这条规则：early ack可能在执行前确认；late ack在工作完成后确认，更便于恢复未确认消息，但完成副作用后、确认前崩溃仍可能重复。不要把任一配置称为exactly once。

at-least-once有前提

完成第 4 步“观察重投窗口并记录限制”后，再回顾这条规则：可用且按要求配置的broker、未确认消息恢复和消费者重试共同支持至少一次交付目标。Celery即使late ack，部分worker子进程丢失场景默认仍确认；reject_on_worker_lost改变该行为，也可能制造毒消息循环。

幂等约束逻辑副作用

完成第 3 步“重复投递同一业务任务”后，再回顾这条规则：同一job_id使用不可变快照和稳定输出名称，重复执行产生相同内容。本地原子替换只避免读到半文件，不解决多个版本竞争、断电持久性或跨机器文件一致性。

技术文档

• Celery任务与ack
• Redis可见性时间
• Python os.replace

---

10 为重试设置退避、抖动与总预算

商品来源暂时不可读，立即判失败太早，永远重试又会拖住任务。用可控失败观察何时再试、何时停止。

从一个问题开始

商品来源暂时不可读，立即判失败太早，永远重试又会拖住任务。用可控失败观察何时再试、何时停止。

这次的终点

前两次暂时失败后恢复成功；已经过期的请求立即失败，永久错误不重试，最多执行首次加三次重试。

开始前，确认你在这里

• 已完成任务第 08–09 课，保留可用 Celery 链路；在 tasks_lab 新建 retry_tasks.py，加入 include 后重启 worker，另开 producer 终端验证。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。
• worker、producer、publisher 均在同一个 Linux/WSL2 环境的 tasks_lab 启动；Bash 用 source .venv/bin/activate。环境变量只在当前终端生效，新终端按本课文件加载 .env.tasks，不复制 Windows 虚拟环境。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 用明确异常类型区分暂时读取失败与永久输入错误
• 在同一截止时间内设置有限重试、退避和随机等待
• 分别验证两次失败后恢复、立即过期，以及最多首次加三次重试

一步步完成案例

1. 定义可控的暂时故障

新建retry_tasks.py，并把该模块加入celery_app.py的include列表。任务只模拟读取，不写文件，便于观察重试而不重复副作用。fail_until只用于本地演练，不对外开放。

文件：retry_tasks.py

import random
import time
from celery_app import app

class TemporaryReadError(Exception):
    pass

def fetch_snapshot(attempt: int, fail_until: int):
    if attempt < fail_until:
        raise TemporaryReadError("temporary source failure")
    return [{"id": 1, "name": "教学样品", "description": "", "is_active": False, "price_cents": 0}]

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。fetch_snapshot 在 attempt 小于 fail_until 时抛 TemporaryReadError，达到阈值则返回一件商品；本步不写导出文件。

刚才发生了什么

先给“过一会可能好”的错误一个明确类型，重试逻辑才能只处理它。参数错误这类无法靠等待修复的问题应直接失败。

2. 加入跨重试截止时间

追加任务。进入执行前和计划下次重试前都检查；每次真实I/O还须设置不超过remaining的超时。手动retry会抛出Retry用于通知Celery，使用raise保持控制流。

文件：retry_tasks.py

@app.task(bind=True, name="products.retry_demo", max_retries=3)
def retry_export(self, job_id: str, deadline_epoch: float, fail_until: int = 2):
    remaining = deadline_epoch - time.time()
    if remaining <= 0:
        raise RuntimeError("export deadline exceeded")
    try:
        rows = fetch_snapshot(self.request.retries, fail_until)
        return {"job_id": job_id, "rows": len(rows)}
    except TemporaryReadError as exc:
        if self.request.retries >= self.max_retries:
            raise
        cap = min(8.0, 2.0 ** self.request.retries)
        delay = random.uniform(0, cap)
        if time.time() + delay + 0.1 >= deadline_epoch:
            raise RuntimeError("insufficient retry budget") from exc
        raise self.retry(exc=exc, countdown=delay)

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。任务函数同时检查截止时间和 max_retries=3；每次计划重试都保留原 deadline_epoch，随机等待不超过本次 cap。

刚才发生了什么

退避让后续尝试逐渐放慢，抖动让多个失败任务错开再试。deadline 是整个任务共用的最后期限，不能每次重试重新获得一整份时间预算。 deadline_epoch由producer确定并原样传递。

3. 分别验证恢复与预算耗尽

重启worker加载模块；在独立producer运行两次。get只用于本地命令行，异常时记录类型。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python verify_retry.py，再与下方预期对照。

文件：verify_retry.py

import time
from uuid import uuid4
from retry_tasks import retry_export

ok = retry_export.delay(str(uuid4()), time.time() + 30, 2)
print(ok.get(timeout=35))
expired = retry_export.delay(str(uuid4()), time.time() - 1, 0)
try:
    expired.get(timeout=10)
except Exception as exc:
    print(type(exc).__name__, str(exc))

你应该看到什么

第一个结果返回 rows=1，worker 可见两次重试后成功；第二个打印 RuntimeError 和 export deadline exceeded。

刚才发生了什么

对比这两个结果可知，次数没用完也可能因时间耗尽停止。get(timeout) 限制的是验收终端等待，业务截止时间则是消息随重试继续携带的约束。 第一次前两次失败后成功；第二次预算已经过期，应在调用外部读取前失败。

4. 验证不能恢复的永久错误

临时把fetch_snapshot改成raise ValueError，确认不会进入TemporaryReadError重试分支。再将fail_until设置10，观察最多初次加3次重试。记录开始时间、重试次数、等待区间和终态；恢复实验函数。

文件：retry_tasks.py

# 永久错误实验：
# raise ValueError("invalid export request")

# 次数预算实验：
# result = retry_export.delay(str(uuid4()), time.time() + 60, 10)
# 最多4次执行：首次 + 3次重试。

你应该看到什么

临时 ValueError 只执行一次；fail_until=10 时最多出现四次执行，之后失败，恢复函数后正常案例继续通过。

刚才发生了什么

重试次数是包含边界的承诺：max_retries=3 不包含首次执行。只捕获 TemporaryReadError，避免把永久错误也变成不断延迟的任务。

检查学习结果

• fail_until=2的任务经历重试后完成。
• 已过deadline的任务不执行读取，直接失败。
• 永久ValueError不重试，最多执行次数为首次加max_retries。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

导出依赖短暂不可用的商品存储，用户只接受10分钟内完成。

实现决策

• 持久任务记录保存deadline和尝试数，不只依赖消息内的request.retries。
• 每个外部操作配置阶段超时，预计等待超过预算就失败。
• 任务副作用保持幂等，避免重试重复发布或重复扣费。

隔离环境演练

对本地函数分别注入可恢复错误、永久错误、过期deadline，记录三种重试轨迹。

恢复验收

• 重试有次数上限和总时间上限。
• 不可恢复错误不会无限排队。
• 已超业务时限的任务不会继续消耗下游。

遇到问题时查这里

每次重试都重置30秒deadline

原因：总预算实际永远不会耗尽。

处理：deadline随原始任务保存，重试传递原值，不在worker重新计算。

对所有Exception自动重试

原因：语法、校验或权限错误变成无用流量。

处理：明确可重试异常集合，永久故障直接进入失败处理。

自己试一次

把预算设为0.05秒，说明为什么次数预算尚余也可能直接失败。

提示：执行机会必须同时满足次数和时间条件。

参考答案：

remaining已不足下一次等待加最小操作时间，因此应结束；剩余retry次数不是必须消耗的额度。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

重试只针对有机会恢复的失败

完成第 1 步“定义可控的暂时故障”后，再回顾这条规则：连接暂不可用、受控限流等可以重试；输入校验失败、无权访问通常直接失败。重试范围应围绕幂等的最小操作，不应重复整个含不可逆副作用的业务流程。

次数与总时间都要限制

完成第 2 步“加入跨重试截止时间”后，再回顾这条规则：max_retries只限制retry次数，不限制首次排队、每次执行和等待的总耗时。deadline_at随业务任务持久保存；跨worker用明确时间基准，本地耗时测量使用monotonic。

退避与抖动减少同步冲击

完成第 2 步“加入跨重试截止时间”后，再回顾这条规则：等待上限随尝试次数增长，再在区间内随机选择等待时间。长远调度不宜大量使用Redis broker的ETA/countdown，避免消息占用worker内存及可见性时间导致重投。

技术文档

• Celery retry与backoff
• Celery Redis ETA约束
• Python时间基准

---

11 用 MySQL 保存任务状态与租约

Celery 的结果缓存到期后，运营还需要知道导出是否成功。把任务记录存入 MySQL，并防止旧执行者覆盖新结果。

从一个问题开始

Celery 的结果缓存到期后，运营还需要知道导出是否成功。把任务记录存入 MySQL，并防止旧执行者覆盖新结果。

这次的终点

任务状态与固定商品快照落库；只传 job_id 即可执行导出，旧令牌与过期租约不能发布成功。

开始前，确认你在这里

• 已完成任务第 08–10 课和数据库模块，shop_lab 中至少有一件商品；在 Linux/WSL2 的 tasks_lab 配置同步 mysql+pymysql URL，先在 MySQL 客户端核对 DATABASE()=shop_lab。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。
• worker、producer、publisher 均在同一个 Linux/WSL2 环境的 tasks_lab 启动；Bash 用 source .venv/bin/activate。环境变量只在当前终端生效，新终端按本课文件加载 .env.tasks，不复制 Windows 虚拟环境。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 将任务状态与固定商品快照保存到 MySQL，而非只依赖结果缓存
• 让 worker 只接收 job_id，领取后生成独立令牌文件并提交结果引用
• 验证重复领取被拒绝、旧令牌失效及新领取次数累加

一步步完成案例

1. 建立隔离状态表

在本机 MySQL 客户端连接 shop_lab，用 app_user 核对当前版本、UTC 和已有 products 表。商品表应保留 id、name、description、is_active、price_cents、version 及 UTC 时间字段，不能覆盖已有数据。把下方 DDL 保存为 schema_jobs.sql，在该库单独执行建表；不放进业务事务。随后执行 SHOW CREATE TABLE export_jobs 核对字段与索引。

文件：schema_jobs.sql

SET time_zone = '+00:00';
SELECT VERSION(), @@session.time_zone, @@session.transaction_isolation;
SHOW CREATE TABLE products;

CREATE TABLE IF NOT EXISTS export_jobs (
  id CHAR(36) CHARACTER SET ascii COLLATE ascii_bin PRIMARY KEY,
  status VARCHAR(16) NOT NULL DEFAULT 'queued',
  payload JSON NOT NULL,
  attempts INT NOT NULL DEFAULT 0,
  lease_token CHAR(36) CHARACTER SET ascii COLLATE ascii_bin,
  lease_until DATETIME(6),
  next_attempt_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  last_dispatched_at DATETIME(6),
  cancel_requested BOOLEAN NOT NULL DEFAULT FALSE,
  deadline_at DATETIME(6) NOT NULL,
  result_ref TEXT,
  error_code VARCHAR(64),
  created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  CONSTRAINT export_jobs_status_check
    CHECK (status IN ('queued','running','succeeded','failed','cancelled')),
  CONSTRAINT export_jobs_attempts_check CHECK (attempts >= 0),
  KEY export_jobs_status_created (status, created_at),
  KEY export_jobs_recovery (status, next_attempt_at, updated_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

你应该看到什么

SHOW CREATE TABLE export_jobs 可见状态、payload、attempts、lease_token/lease_until、deadline 和恢复索引；products 保持原数据。

刚才发生了什么

payload 留住本次清单输入，status 留住进度。租约是“在某个期限前可处理这项任务”的权利，令牌标识本次持有者；attempts 记录真实领取次数。

2. 在短事务中固定导出快照

在 tasks_lab 中创建 state_store.py；第 08 课已安装 PyMySQL[rsa]，当前 WSL 虚拟环境若缺少则补装。将 DATABASE_URL=mysql+pymysql://app_user:已百分号编码的本机密码@127.0.0.1:3307/shop_lab 写入受保护的 .env.tasks，不打印连接串，也不改用旧认证插件。下面代码按 connect、transaction、快照读取与提交顺序保存。本步只定义函数，第 5 步才运行提交实验；代码中的参数绑定、commit/rollback/finally 都要保留。提交应答丢失时不猜测结果，不无界重试。

文件：state_store.py

import json
import os
import re
from contextlib import contextmanager
from urllib.parse import parse_qs, unquote, urlsplit
from uuid import uuid4
import pymysql
from pymysql.cursors import DictCursor
from dotenv import load_dotenv

load_dotenv(".env.tasks")


def decode_url_part(value: str) -> str:
    if re.search(r"%(?![0-9A-Fa-f]{2})", value):
        raise ValueError("invalid percent encoding")
    return unquote(value, encoding="utf-8", errors="strict")


def connect():
    parts = urlsplit(os.environ["DATABASE_URL"])
    options = parse_qs(parts.query, keep_blank_values=True, strict_parsing=True)
    if (parts.scheme != "mysql+pymysql" or not parts.hostname
            or not parts.username or parts.password is None or parts.fragment):
        raise ValueError("expected a mysql+pymysql database URL")
    if options and options != {"charset": ["utf8mb4"]}:
        raise ValueError("only charset=utf8mb4 is supported in this local URL")
    database = decode_url_part(parts.path.removeprefix("/"))
    if not re.fullmatch(r"[A-Za-z0-9_]{1,64}", database):
        raise ValueError("invalid database name")
    conn = pymysql.connect(
        host=parts.hostname, port=parts.port or 3306,
        user=decode_url_part(parts.username),
        password=decode_url_part(parts.password), database=database,
        charset="utf8mb4", cursorclass=DictCursor, autocommit=False,
        connect_timeout=3, read_timeout=3, write_timeout=3,
    )
    try:
        with conn.cursor() as cur:
            cur.execute("SET time_zone = '+00:00'")
            cur.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED")
            cur.execute("SET SESSION innodb_lock_wait_timeout = 1")
            cur.execute("SET SESSION max_execution_time = 3000")
        conn.commit()
    except BaseException:
        try:
            conn.rollback()
        except pymysql.MySQLError:
            pass
        conn.close()
        raise
    return conn


@contextmanager
def transaction():
    conn = connect()
    try:
        with conn.cursor() as cur:
            yield cur
        conn.commit()
    except BaseException:
        try:
            conn.rollback()
        except pymysql.MySQLError:
            pass
        raise
    finally:
        conn.close()


def decode_payload(raw):
    value = json.loads(raw) if isinstance(raw, (str, bytes, bytearray)) else raw
    if not isinstance(value, list) or not all(isinstance(row, dict) for row in value):
        raise ValueError("expected a product snapshot array")
    return value


def read_snapshot(cur, product_ids: list[int]):
    if not isinstance(product_ids, list) or not product_ids or len(product_ids) > 100:
        raise ValueError("select 1..100 products")
    if any(type(product_id) is not int or product_id <= 0 for product_id in product_ids):
        raise ValueError("product IDs must be positive integers")
    ids = sorted(set(product_ids))
    placeholders = ",".join(["%s"] * len(ids))
    cur.execute(
        f"SELECT id,name,description,is_active,price_cents FROM products WHERE id IN ({placeholders}) ORDER BY id",
        tuple(ids),
    )
    rows = cur.fetchall()
    if len(rows) != len(ids):
        raise ValueError("one or more products do not exist")
    for row in rows:
        row["is_active"] = bool(row["is_active"])
        row["price_cents"] = int(row["price_cents"])
    return rows


def submit_snapshot(product_ids: list[int]) -> str:
    job_id = str(uuid4())
    with transaction() as cur:
        rows = read_snapshot(cur, product_ids)
        cur.execute(
            """INSERT INTO export_jobs(id,payload,deadline_at)
               VALUES (%s,%s,DATE_ADD(NOW(6), INTERVAL 10 MINUTE))""",
            (job_id, json.dumps(rows, ensure_ascii=False)),
        )
        if cur.rowcount != 1:
            raise RuntimeError("job insert did not create one row")
    return job_id

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。调用 submit_snapshot 返回 UUID，数据库出现 queued 行与固定商品数组；无效或不存在编号会失败，不留下部分任务记录。

刚才发生了什么

transaction 把读取快照和插入任务包成一个提交单位。快照让重试读取同一份名称、价格和上架状态，而不是随着商品编辑改变导出内容。 会话设 UTC 与 READ COMMITTED，使队列和容量读看到每条语句开始时最新已提交数据。 connect 返回原生连接，transaction 交出能按字段名读结果的 DictCursor；正常退出提交，异常回滚，最后关闭。数据库连接统一 UTC 与 READ COMMITTED，读取到的无时区 DATETIME 按已约定的 UTC 解释。连接与 socket 读写限 3 秒、行锁等待 1 秒；max_execution_time 只限制适用的只读 SELECT，并不是所有操作的硬期限。

3. 添加领取与完成的CAS

追加领取与完成函数。先在短事务中 SELECT ... FOR UPDATE SKIP LOCKED 锁定可领取任务，再 CAS 更新；提交后才执行文件操作。取不到行也可能只是被另一事务暂时锁住，不等于任务不存在。running 租约过期也不能直接领取，由第 15 课恢复为 queued 并持久写入 10 秒退避；所有重复消息都必须经过 next_attempt_at。完成用 rowcount 判断 CAS，30 秒租约仅适用于本地小文件；长导出需分块续租并停止失去权属后的发布。

文件：state_store.py

def claim(job_id: str):
    token = str(uuid4())
    with transaction() as cur:
        cur.execute("""
            SELECT id FROM export_jobs
            WHERE id=%s AND status='queued' AND cancel_requested=FALSE
              AND attempts<3 AND deadline_at>NOW(6) AND next_attempt_at<=NOW(6)
            FOR UPDATE SKIP LOCKED
        """, (job_id,))
        if cur.fetchone() is None:
            return None
        cur.execute("""
            UPDATE export_jobs SET status='running', attempts=attempts+1,
              lease_token=%s, lease_until=DATE_ADD(NOW(6), INTERVAL 30 SECOND),
              error_code=NULL, updated_at=NOW(6)
            WHERE id=%s AND status='queued' AND cancel_requested=FALSE
              AND attempts<3 AND deadline_at>NOW(6) AND next_attempt_at<=NOW(6)
        """, (token, job_id))
        if cur.rowcount != 1:
            return None
        cur.execute(
            "SELECT id,payload,lease_token,attempts FROM export_jobs WHERE id=%s",
            (job_id,),
        )
        row = cur.fetchone()
        if row is None or row["lease_token"] != token:
            raise RuntimeError("claim ownership changed inside a locked transaction")
        row["payload"] = decode_payload(row["payload"])
    return row


def finish(job_id: str, token: str, result_ref: str) -> bool:
    with transaction() as cur:
        cur.execute("""
            UPDATE export_jobs SET status='succeeded', result_ref=%s,
              error_code=NULL, updated_at=NOW(6)
            WHERE id=%s AND status='running' AND lease_token=%s
              AND lease_until>NOW(6) AND deadline_at>NOW(6)
              AND cancel_requested=FALSE
        """, (result_ref, job_id, token))
        changed = cur.rowcount == 1
    return changed

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。同一 queued 任务第一次 claim 得到 token、payload 和 attempts=1，未结束时再次 claim 返回 None；finish 必须匹配当前 token。

刚才发生了什么

先锁行再更新，能让两个 worker 竞争时只由一个领取。CAS 是“条件仍符合才修改”：完成也检查令牌、期限和取消状态，防止已经失去权利的旧执行者写回成功。 PyMySQL 的 JSON 列通常返回字符串，领取时用 decode_payload 转为列表。

4. 替换worker为只接收job_id

整体替换export_tasks.py并重启worker，停止使用第08–09课携带rows的producer。每次领取写独立token文件，最终有效文件由数据库result_ref决定；旧worker最多留下未引用文件，不覆盖新结果。异常由Celery记录，业务状态暂留租约，恢复扫描在第15课补齐。不要把第10课retry直接套在未定义状态转换上。

文件：export_tasks.py

import json
from pathlib import Path
from celery_app import app
from state_store import claim, finish

@app.task(name="products.export")
def export_products(job_id: str):
    job = claim(job_id)
    if job is None:
        return {"job_id": job_id, "claimed": False}
    token = str(job["lease_token"])
    folder = Path("artifacts")
    folder.mkdir(exist_ok=True)
    target = folder / f"{job['id']}.{token}.json"
    target.write_text(json.dumps(job["payload"], ensure_ascii=False), encoding="utf-8")
    published = finish(job_id, token, str(target))
    return {"job_id": job_id, "published": published}

你应该看到什么

新 worker 只接受一个 job_id；领取成功后写 job_id.token.json，finish 成功时数据库 result_ref 指向它。

刚才发生了什么

消息只负责叫醒执行者，真正输入从持久状态表读取。每次令牌使用独立文件，旧执行者最多留下未引用文件，数据库指向的结果才是有效结果。

5. 从数据库查询并测试旧token失效

先从 products 取真实存在的编号，再执行 submit_snapshot 和 export_products.delay。transaction 即使只读也显式结束事务并关闭连接。当前提交后再发消息仍有双写窗口，下一课用 outbox 补齐。HTTP 适配层需另行接入这份状态，不能把第 01 课内存字典当数据库状态。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python submit_db_export.py，再与下方预期对照。

文件：submit_db_export.py

from state_store import submit_snapshot, transaction
from export_tasks import export_products

with transaction() as cur:
    cur.execute("SELECT MIN(id) AS id FROM products")
    product_id = cur.fetchone()["id"]
assert product_id is not None, "create a product first"
job_id = submit_snapshot([product_id])
export_products.delay(job_id)
with transaction() as cur:
    cur.execute(
        "SELECT id,status,attempts,result_ref FROM export_jobs WHERE id=%s",
        (job_id,),
    )
    print(cur.fetchone())

# 持久 HTTP 查询适配层也应读同一行；无记录才返回 404。
# 只针对本次 job_id 进行后面的租约实验。

你应该看到什么

提交后查询同一行，开始可能为 queued/running，重复只读查询最终为 succeeded 且 result_ref 非空；读取实际文件能看到快照。

刚才发生了什么

消息发送后 worker 独立执行，立即查询不保证已经完成。数据库行成为跨进程、跨结果缓存过期的查询依据；提交后再发送仍有一步漏发窗口，下一课处理。

6. 只对专用任务验证过期与旧令牌

使用已有商品编号作为输入；下面通过最小编号选取样本，不假定id=1存在。只修改刚创建的job_id。过期running无法自动领取，实验显式模拟第15课的恢复转换；attempts不清零，随后旧token仍不能完成，新token生成真实文件后才发布。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python verify_lease.py，再与下方预期对照。

文件：verify_lease.py

import json
from pathlib import Path
from state_store import submit_snapshot, transaction, claim, finish

with transaction() as cur:
    cur.execute("SELECT MIN(id) AS id FROM products")
    product_id = cur.fetchone()["id"]
assert product_id is not None, "create a product first"
job_id = submit_snapshot([product_id])
old = claim(job_id)
assert old is not None and claim(job_id) is None
with transaction() as cur:
    cur.execute(
        "UPDATE export_jobs SET lease_until=DATE_SUB(NOW(6), INTERVAL 1 SECOND) WHERE id=%s",
        (job_id,),
    )
assert claim(job_id) is None
assert finish(job_id, old["lease_token"], "unused") is False
with transaction() as cur:
    cur.execute("""
        UPDATE export_jobs SET status='queued',lease_token=NULL,lease_until=NULL,
          next_attempt_at=NOW(6),updated_at=NOW(6)
        WHERE id=%s AND status='running' AND lease_until<NOW(6)
    """, (job_id,))
    assert cur.rowcount == 1
new = claim(job_id)
assert new is not None and new["attempts"] == 2
assert finish(job_id, old["lease_token"], "unused") is False
folder = Path("artifacts")
folder.mkdir(exist_ok=True)
target = folder / f"{job_id}.{new['lease_token']}.json"
target.write_text(json.dumps(new["payload"],ensure_ascii=False),encoding="utf-8")
assert finish(job_id,new["lease_token"],str(target)) is True
print("lease ownership verified",job_id)

你应该看到什么

脚本所有断言通过：过期任务不能直接重领，旧 token 两次 finish 都为 False，新领取 attempts=2，真实新文件发布成功。

刚才发生了什么

租约过期不是授予旧进程更多时间。先做明确恢复转换，再发新令牌，完成时仍检查所有权，才能让晚到的旧写回失效。

检查学习结果

• 任务记录重启后仍存在，payload是提交时固定的商品快照。
• 同一租约期内重复领取只有一方取得token。
• 旧token无法完成任务，工作期间不持有数据库长事务。
• next_attempt_at未到时所有重复消息都无法领取；过期running由恢复扫描转换后才能再次领取。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

多个worker可能收到同一个导出job，旧worker在暂停后重新运行。

实现决策

• 业务job与Celery消息分开，任务状态保存在MySQL。
• 输入快照固定；多租约输出隔离，CAS只发布当前token结果。
• 授权过滤应复用shop-service当前用户权限，本地单账户练习不能直接部署为多用户接口。
• finish同时检查业务deadline；超时后不能仅凭未过租约发布成功。
• 领取资格来自数据库状态与next_attempt_at，不由broker是否再次交付决定。
• 连接、socket、行锁和 SELECT 超时分别配置；连接中断或提交应答丢失不能证明回滚。未知结果先按 job_id 核对，外层重试必须保留总预算。

隔离环境演练

对专用任务令租约过期，先确认claim和旧token完成均失败，再模拟恢复为queued并重新领取，比较两次token与result_ref。

恢复验收

• 旧执行者不能发布结果。
• 数据库事务不跨耗时文件处理。
• attempts与deadline来自持久记录，不因消息重投重置。

遇到问题时查这里

旧worker覆盖新worker结果

原因：完成时只按job_id更新，没有检查token或输出路径复用。

处理：文件名带token，结果发布使用状态、token和租约条件CAS。

Redis显示SUCCESS但数据库仍queued或running

原因：消息可能没有领取权，或者完成CAS失败；执行结果不等于业务终态。

处理：用户查询只认export_jobs，记录claimed/published用于诊断。

自己试一次

实现续租函数：仅当token匹配、status=running、当前租约未过期且未取消时延长30秒。

提示：过期租约不能自己复活，需要重新领取获取新token。

参考答案：

在 transaction() 中执行 UPDATE export_jobs SET lease_until=LEAST(DATE_ADD(NOW(6), INTERVAL 30 SECOND),deadline_at) WHERE id=%s AND lease_token=%s AND status='running' AND lease_until>NOW(6) AND deadline_at>NOW(6) AND cancel_requested=FALSE；参数是 (job_id,token)。用 rowcount 与当前行状态核对：重复到同一截止值可不产生变化，不能因此把仍有效的租约误报失效；需在同一短事务继续 SELECT 校验有效 token、租约和 deadline。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

状态表是用户查询真源

完成第 5 步“从数据库查询并测试旧token失效”后，再回顾这条规则：MySQL保存输入快照、状态、尝试数、截止时间和结果引用。Redis结果过期不改变业务任务是否存在。此后Celery消息只传job_id，不再传第08课rows参数。

领取与完成都要条件更新

完成第 3 步“添加领取与完成的CAS”后，再回顾这条规则：领取先锁定已到 next_attempt_at 的 queued 行，再用条件 UPDATE 设置 lease_token 并增加 attempts，两步位于同一短事务。完成检查 running、token、租约、deadline 和取消标志。过期 running 必须经恢复扫描重新排队并设置退避，重复消息不能直接抢占。

短事务包围状态转换

完成第 2 步“在短事务中固定导出快照”后，再回顾这条规则：读取快照和领取状态分别在短事务中完成，文件生成放在事务外。不要持有数据库行锁等待长await、网络上传或CPU计算。租约是有时限的权属，不是自动杀死旧worker。

技术文档

• MySQL 8.4 锁定读取
• MySQL 8.4 UPDATE
• PyMySQL 连接与事务
• MySQL 8.4 事务隔离级别
• PyMySQL 参数绑定与提交
• MySQL 8.4 SELECT 超时边界
• PyMySQL 认证依赖

---

12 用事务 outbox 连接状态提交与消息发布

任务已经记到 MySQL，发送消息时却断网。上一课两步之间可能漏掉执行通知，现在先把“待发送”也一起记下来。

从一个问题开始

任务已经记到 MySQL，发送消息时却断网。上一课两步之间可能漏掉执行通知，现在先把“待发送”也一起记下来。

这次的终点

提交时同时保存任务和待发事件；broker 暂时不可用时事件保留，重复发布仍只保留一个有效业务结果。

开始前，确认你在这里

• 已完成任务第 11 课，state_store.py 的 submit_snapshot/claim/finish 可用；tasks_lab 中 Celery 配置仍指向 6384，MySQL 客户端确认当前库 shop_lab。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。
• worker、producer、publisher 均在同一个 Linux/WSL2 环境的 tasks_lab 启动；Bash 用 source .venv/bin/activate。环境变量只在当前终端生效，新终端按本课文件加载 .env.tasks，不复制 Windows 虚拟环境。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 让任务记录与待发事件在同一数据库事务提交
• 在事务之外发送消息，随后按当前令牌确认 sent_at
• 模拟发送后未确认和 broker 不可用，核对事件保留与重复发布结果

一步步完成案例

1. 建立outbox表

在同一 shop_lab 数据库的 UTC 会话执行。外键会立即检查 job 已存在，所以提交事务先 INSERT job，再 INSERT outbox；二者仍在同一事务提交。事件 ID 与业务 ID 分开，同一个 job 后续可以有恢复事件。MySQL 不支持带 WHERE 条件的部分索引，这里使用 sent_at、available_at、created_at 联合索引，并为按 job 查询事件建立索引。

文件：schema_outbox.sql

CREATE TABLE IF NOT EXISTS export_outbox (
  id CHAR(36) CHARACTER SET ascii COLLATE ascii_bin PRIMARY KEY,
  job_id CHAR(36) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
  created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  available_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  sent_at DATETIME(6),
  claim_token CHAR(36) CHARACTER SET ascii COLLATE ascii_bin,
  claim_until DATETIME(6),
  attempts INT NOT NULL DEFAULT 0,
  CONSTRAINT export_outbox_job_fk FOREIGN KEY (job_id) REFERENCES export_jobs(id),
  KEY export_outbox_pending (sent_at, available_at, created_at),
  KEY export_outbox_job_sent (job_id, sent_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

你应该看到什么

export_outbox 存在，job_id 外键指向 export_jobs；sent_at、领取令牌、期限和两个索引均可由 SHOW CREATE TABLE 核对。

刚才发生了什么

outbox 是数据库中的待发清单。任务编号指向要执行的业务，事件编号指向一次发送记录；分开后同一任务还能产生恢复通知。

2. 让提交记录与事件一起提交

用下方完整函数替换 tasks_lab/state_store.py 中第 11 课的 submit_snapshot，保留其他 helper。状态和事件使用同一个 transaction，任一步异常都会回滚。此步只改函数，下一步将整体替换并运行第 11 课的 submit_db_export.py，删除它原先导入 export_tasks 与直接 export_products.delay(job_id) 的两处代码。HTTP 适配层只能在函数成功返回、事务提交之后回应 202；publisher 的网络发送不能放入本事务。

文件：state_store.py

# 替换 state_store.py 中的同名函数；复用已定义的 read_snapshot / transaction。
def submit_snapshot(product_ids: list[int]) -> str:
    job_id = str(uuid4())
    with transaction() as cur:
        rows = read_snapshot(cur, product_ids)
        cur.execute(
            """INSERT INTO export_jobs(id,payload,deadline_at)
               VALUES (%s,%s,DATE_ADD(NOW(6), INTERVAL 10 MINUTE))""",
            (job_id, json.dumps(rows, ensure_ascii=False)),
        )
        if cur.rowcount != 1:
            raise RuntimeError("job insert did not create one row")
        cur.execute(
            "INSERT INTO export_outbox(id,job_id) VALUES (%s,%s)",
            (str(uuid4()), job_id),
        )
        if cur.rowcount != 1:
            raise RuntimeError("outbox insert did not create one row")
    return job_id

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。替换后只调用 submit_snapshot 会同时增加 job 与未发送 outbox，尚不直接调用 Celery；任一插入失败两者都回滚。

刚才发生了什么

把两条写入放在一个事务里，解决“任务保存了，通知却没留下”的间隙。提交成功表示可靠受理，消息何时送达由发布器继续负责。

3. 提交一个带待发事件的新任务

先暂停本机已经运行的 publisher（首次学习此课尚未建立发布器则无需停止），避免新事件立即被取走。用下面完整代码替换第 11 课的 tasks_lab/submit_db_export.py，再在 tasks_lab 终端运行 python submit_db_export.py。保存输出中的本次 job_id；本脚本只提交 MySQL，不直接调用 Celery。下一步创建发布器，发送前核对这里创建的事件仍未发送。

文件：submit_db_export.py

from state_store import submit_snapshot, transaction

with transaction() as cur:
    cur.execute('SELECT MIN(id) AS id FROM products')
    product_id = cur.fetchone()['id']
assert product_id is not None, 'create a product first'
job_id = submit_snapshot([product_id])
with transaction() as cur:
    cur.execute("""SELECT j.id,j.status,o.id AS event_id,o.sent_at
                   FROM export_jobs j JOIN export_outbox o ON o.job_id=j.id
                   WHERE j.id=%s""", (job_id,))
    row = cur.fetchone()
    assert row is not None and row['status'] == 'queued' and row['sent_at'] is None
    print(row)
print('ready for publisher:', job_id)

你应该看到什么

打印本次 job_id、queued、独立 event_id 与 sent_at=None，并出现 ready for publisher；商品任务与未发事件共同存在，尚未产生结果文件。

刚才发生了什么

保存函数不会自动调用它，这一步真正创建一份要执行的工作。取消提交端的直接 delay 后，唯一发送入口成为读取 outbox 的发布器，因此能亲眼确认“已受理但还未发消息”这个持久阶段。

4. 领取一条待发布事件

创建 publisher.py。READ COMMITTED 下先锁定一条候选事件，再在同一短事务写领取 token，返回前已提交并释放锁。MySQL 的 LIMIT 位于 FOR UPDATE SKIP LOCKED 之前；SKIP LOCKED 只用于允许跳过繁忙记录的工作队列，不适用于普通一致性报表。一次只处理一条，部署循环需空批等待、停机检查和异常告警。

文件：publisher.py

from uuid import uuid4
from state_store import transaction
from celery_app import app


def claim_event():
    token = str(uuid4())
    with transaction() as cur:
        cur.execute("""
            SELECT id,job_id FROM export_outbox
            WHERE sent_at IS NULL AND available_at<=NOW(6)
              AND (claim_until IS NULL OR claim_until<NOW(6))
            ORDER BY created_at,id LIMIT 1 FOR UPDATE SKIP LOCKED
        """)
        event = cur.fetchone()
        if event is None:
            return None
        cur.execute("""
            UPDATE export_outbox SET claim_token=%s,
              claim_until=DATE_ADD(NOW(6), INTERVAL 30 SECOND),attempts=attempts+1
            WHERE id=%s AND sent_at IS NULL AND available_at<=NOW(6)
              AND (claim_until IS NULL OR claim_until<NOW(6))
        """, (token,event["id"]))
        if cur.rowcount != 1:
            raise RuntimeError("outbox claim no longer matched")
        event["claim_token"] = token
    return event

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。claim_event 一次最多返回一条事件及新 claim_token；领取事务结束后能在库中看到 30 秒 claim_until。

刚才发生了什么

先短暂锁住候选行再写领取权，其他发布器可跳过忙行。函数返回时锁已释放，接下来等网络不会一直占用数据库事务。

5. 发送后按token确认

把代码追加到 publisher.py，确认本课第 3 步打印的专用任务仍有 sent_at=NULL 的事件，再运行 python publisher.py。若还有其他实验的待发事件，单次调用可能先发送它们，应按保存的 job_id 查询对应 outbox 并重复有界单次发布，直到本次事件 sent_at 非空。发送失败不标 sent，领取租约到期后再尝试；rowcount=0 不能宣称确认成功。第 15 课还会补充同事务派发时间与未领取恢复，发布失败需记录并保持运行预算。

文件：publisher.py

def publish_one():
    event = claim_event()
    if event is None:
        return False
    app.send_task(
        "products.export",args=[event["job_id"]],
        task_id=event["id"],queue="exports",retry=False,
    )
    with transaction() as cur:
        cur.execute("""
            UPDATE export_outbox SET sent_at=NOW(6),claim_until=NULL
            WHERE id=%s AND claim_token=%s AND sent_at IS NULL
              AND claim_until>NOW(6)
        """, (event["id"],event["claim_token"]))
        confirmed = cur.rowcount == 1
    return confirmed


if __name__ == "__main__":
    print("published",publish_one())

你应该看到什么

有待发事件时 python publisher.py 打印 published True，sent_at 写入；无可领取事件为 False，发送失败则抛错且不标 sent。

刚才发生了什么

网络发送和数据库确认是两次动作，不能合成一个普通事务。只有当前 token 和有效租约匹配时才确认，避免旧发布器覆盖别人接管后的状态。

6. 注入发送后未标记的故障

先再次运行本课第 3 步的 python submit_db_export.py，建立一项新的 queued 任务与未发事件并保存 job_id。只在本机 publisher 临时跳过发送后的数据库确认更新，然后运行 python publisher.py；按本次 job_id 确认事件 sent_at 仍为空。等其 30 秒领取租约到期、恢复确认代码后再次发布，核对 export_jobs 只有一个有效 result_ref。另试暂停专用 broker 后再提交新任务，确认 job/outbox 同时保留，恢复 broker 后再发布。

文件：终端 / 当前课时

SELECT j.id, j.status, j.result_ref, o.id AS event_id, o.attempts, o.sent_at
FROM export_jobs j JOIN export_outbox o ON o.job_id=j.id
ORDER BY o.created_at DESC LIMIT 10;

你应该看到什么

跳过确认后 sent_at 仍为空，租约过期重跑时 attempts 增加；最终任务只有一个有效 result_ref，停止 broker 后提交仍保存 job/outbox。

刚才发生了什么

这次重发正是 outbox 的边界：它让漏发可恢复，但允许同一事件发两次。第 11 课的业务状态条件更新负责保护最终有效结果。

检查学习结果

• 提交事务提交后，job与outbox事件同时存在。
• broker不可用时sent_at保持NULL，事件可被后续处理。
• 重复发布不会让旧worker越过租约CAS覆盖终态结果。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

数据库正常、Redis暂不可用，API不能受理后永久漏发任务。

实现决策

• 数据库提交的发送意图与业务记录同生共灭。
• publisher网络调用有超时和重试策略，长期未发布事件报警。
• 重复消息是常规故障路径，consumer继续使用持久幂等与租约。

隔离环境演练

分别模拟事务回滚、broker不可用、发送成功但sent_at未写回，检查三种窗口的恢复结果。

恢复验收

• 没有只存在job而无发送意图的已受理任务。
• 没有跨broker等待的数据库长事务。
• 重复发布可恢复且最终业务结果唯一可定位。

遇到问题时查这里

outbox行已标sent但broker根本没收到

原因：先标记再发送，或把异常忽略。

处理：发送成功后再条件更新sent_at，失败保留待处理事件。

发送很慢导致数据库锁一直占用

原因：把broker发布放在FOR UPDATE事务内部。

处理：分成领取、网络发送、确认三个阶段，用领取token防旧publisher误确认。

自己试一次

在导出记录INSERT后、outbox INSERT前抛异常，验证两个表都没有这次提交的记录。

提示：使用同一个连接事务，让异常退出with触发回滚。

参考答案：

保存本次job_id进行查询；两表计数都应为0。若只回滚一张表，说明事务边界没有统一。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

两个系统之间存在双写窗口

完成第 2 步“让提交记录与事件一起提交”后，再回顾这条规则：先写数据库后发送消息，可能留下永不入队的queued任务；先发送后提交数据库，worker可能读不到记录。outbox把业务记录和发送意图放进同一数据库事务。

outbox不提供exactly once发布

完成第 6 步“注入发送后未标记的故障”后，再回顾这条规则：发布成功后进程可能在写sent_at之前崩溃，事件会被再次发布。consumer继续依赖第11课的领取和结果CAS，不能因为用了outbox就移除幂等逻辑。

发布期间不持数据库事务

完成第 4 步“领取一条待发布事件”后，再回顾这条规则：先短事务领取outbox事件，提交后访问broker，最后短事务按claim_token标记sent。租约过期允许其他publisher接手；SKIP LOCKED用于队列式领取，不用于普通业务一致性查询。

技术文档

• MySQL 8.4 事务提交与回滚
• MySQL 8.4 SELECT
• Celery消息发布
• MySQL 8.4 DDL 隐式提交

---

13 限制受理量并拆分长短任务队列

短清单与长清单挤在一起时，短任务也要排很久；继续无限受理会让用户越等越久。给入口设容量，并为两类任务分开安排执行者。

从一个问题开始

短清单与长清单挤在一起时，短任务也要排很久；继续无限受理会让用户越等越久。给入口设容量，并为两类任务分开安排执行者。

这次的终点

短清单进入 fast、长清单进入 bulk；两个受限 worker 各自消费，容量满时受理接口给出可重试的 503。

开始前，确认你在这里

• 已完成任务第 12 课，publisher.py 可领取并发送 outbox；在 shop_lab 核对现有 queued/running 数量，先处理完旧 exports 队列的实验消息。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。
• worker、producer、publisher 均在同一个 Linux/WSL2 环境的 tasks_lab 启动；Bash 用 source .venv/bin/activate。环境变量只在当前终端生效，新终端按本课文件加载 .env.tasks，不复制 Windows 虚拟环境。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 在同一受理事务中锁定容量门闩、检查总量并新增任务
• 把短清单和长清单分别发送给 fast 与 bulk 的独立消费者
• 达到容量上限时返回 503 与 Retry-After，恢复后观察积压减少

一步步完成案例

1. 建立容量门闩并增加队列字段

在 MySQL 8.4 隔离库的 UTC 会话单独执行一次。DDL 会隐式提交，因此不与业务提交混用。重复学习先 SHOW CREATE TABLE 核对已有表和 queue_name；本迁移不是可无限重跑脚本，不通过忽略所有 SQL 错误掩盖缺失配置。

文件：schema_capacity.sql

SET time_zone = '+00:00';
CREATE TABLE export_admission_gate (
  id TINYINT PRIMARY KEY,
  CONSTRAINT export_admission_singleton CHECK (id=1)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
INSERT INTO export_admission_gate(id) VALUES (1);
COMMIT;

ALTER TABLE export_outbox
  ADD COLUMN queue_name VARCHAR(32) NOT NULL DEFAULT 'exports.bulk',
  ADD CONSTRAINT export_outbox_queue_check
    CHECK (queue_name IN ('exports.fast','exports.bulk'));

你应该看到什么

容量表有且只有 id=1 一行，outbox 新增 queue_name 且仅允许 exports.fast/exports.bulk；旧记录有 bulk 默认值。

刚才发生了什么

这条固定记录充当排队受理的门闩。大家先锁同一行再数容量，避免两个提交同时看到“还剩一位”却都加入。

2. 在受理事务内检查容量

完成上一项 SQL 初始化后，再追加 Python helper。容量门闩是独立 InnoDB 表的一条固定记录：所有提交、周期创建和失败重放都先 FOR UPDATE 锁住 id=1，计数与 INSERT 完成后一起提交。READ COMMITTED 避免在等待门闩前形成旧快照；不能把 MySQL 会话级命名锁当事务门闩。所有提交路径必须使用相同 limit，默认 100；压力实验临时调小额度时必须同时更新所有路径。

文件：state_store.py

# 将以下类和函数追加到 state_store.py。
class CapacityExceeded(Exception):
    pass


def lock_admission(cur):
    cur.execute("SELECT id FROM export_admission_gate WHERE id=1 FOR UPDATE")
    if cur.fetchone() is None:
        raise RuntimeError("initialize the admission gate before submitting")


def admit_export(cur, limit: int = 100):
    if type(limit) is not int or limit < 1:
        raise ValueError("limit must be a positive integer")
    lock_admission(cur)
    cur.execute("""
        SELECT COUNT(*) AS total FROM export_jobs
        WHERE status IN ('queued','running')
    """)
    if cur.fetchone()["total"] >= limit:
        raise CapacityExceeded("export capacity reached")

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。admit_export 在待执行总数达到 limit 时抛 CapacityExceeded；小于上限时在同一事务继续插入。

刚才发生了什么

锁住门闩后检查总量，把检查与新增串成一个整体。限制正在执行的 worker 数量不会自动限制排队量，所以入口需要自己的容量检查。

3. 把事件路由写进outbox

完成本课第一个步骤的迁移后，替换完整 submit_snapshot，避免漏掉容量门闩或 outbox。把 publisher 领取 SELECT 加上 queue_name，并将 send_task 的显式 queue 改为 event['queue_name']；仅修改装饰器不能覆盖旧的显式队列参数。

文件：state_store.py

def submit_snapshot(product_ids: list[int]) -> str:
    job_id = str(uuid4())
    with transaction() as cur:
        admit_export(cur)
        rows = read_snapshot(cur, product_ids)
        cur.execute(
            """INSERT INTO export_jobs(id,payload,deadline_at)
               VALUES (%s,%s,DATE_ADD(NOW(6), INTERVAL 10 MINUTE))""",
            (job_id,json.dumps(rows,ensure_ascii=False)),
        )
        if cur.rowcount != 1:
            raise RuntimeError("job insert did not create one row")
        queue_name = "exports.fast" if len(rows) <= 30 else "exports.bulk"
        cur.execute(
            "INSERT INTO export_outbox(id,job_id,queue_name) VALUES (%s,%s,%s)",
            (str(uuid4()),job_id,queue_name),
        )
        if cur.rowcount != 1:
            raise RuntimeError("outbox insert did not create one row")
    return job_id

# publisher.py 的 claim_event SELECT 改成：
# SELECT id,job_id,queue_name FROM export_outbox ...
# 其余领取条件不变；event 会携带 queue_name。
# publish_one 的 send_task 参数改成 queue=event["queue_name"]。

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。新提交 ≤30 条的事件 queue_name 为 exports.fast，更多为 exports.bulk；publisher 返回事件中也带队列并按该值发送。

刚才发生了什么

路由信息跟事件一起持久保存，重发仍知道应该去哪里。只改 worker 或装饰器不够，显式发送参数决定消息实际进入的队列。 ≤30 条走 fast，其余走 bulk。此步骤复用 state_store 已有 import/helper。

4. 分别启动两个受限worker

先完成第12课旧exports队列中的实验任务，再停旧worker。两个命令分别放在终端运行，各一个执行进程；worker_prefetch_multiplier=1继续保留。正常发布的事件必须进入其中一条被消费的队列。

文件：终端 / 当前课时

# Linux / WSL2，终端A
python -m celery -A celery_app:app worker -l INFO -Q exports.fast --concurrency=1 -n fast@%h

# Linux / WSL2，终端B
python -m celery -A celery_app:app worker -l INFO -Q exports.bulk --concurrency=1 -n bulk@%h

你应该看到什么

两个终端的 worker 日志分别显示 fast@… 和 bulk@…，各只消费自己的队列且 concurrency=1。

刚才发生了什么

队列拆分只有配上独立消费者容量才有效。让长任务占用 bulk 的一个执行位时，fast 的执行位仍可处理短任务。

5. 用数据库观察积压与最老等待

暂停本地消费者后少量提交，逐渐达到本课100条容量；CapacityExceeded应由API转换为503并附Retry-After，而不是继续受理。先统计之前练习遗留的queued/running，不能假定数据库为空。恢复worker和publisher后观察最老等待下降。

文件：终端 / 当前课时

SELECT status,COUNT(*) AS jobs,
       TIMESTAMPDIFF(MICROSECOND,MIN(created_at),NOW(6))/1000000 AS oldest_age_seconds
FROM export_jobs
WHERE status IN ('queued','running')
GROUP BY status;

SELECT queue_name,COUNT(*) AS unpublished,
       TIMESTAMPDIFF(MICROSECOND,MIN(created_at),NOW(6))/1000000 AS oldest_unpublished_seconds
FROM export_outbox WHERE sent_at IS NULL GROUP BY queue_name;

你应该看到什么

暂停消费者后 queued/running 接近上限，新提交被拒绝；恢复发布和消费后积压总量下降，最老等待逐步消化。

刚才发生了什么

待处理数量告诉我们堆了多少，最老年龄告诉我们有人等了多久。之前练习的未完成行也占容量，不能把本节提交次数直接当成全部积压。

6. 给HTTP受理失败返回明确反馈

将CapacityExceeded映射放在现有提交接口，不创建第二个状态真源。不同用户配额用429或业务约定，整个服务容量不足本课使用503。数据库事务回滚后不得返回202和虚假job_id。

文件：admission_adapter.py

from fastapi import HTTPException
from state_store import CapacityExceeded, submit_snapshot

def accept_export(product_ids: list[int]):
    try:
        return {"job_id": submit_snapshot(product_ids)}
    except CapacityExceeded as exc:
        raise HTTPException(
            503, "导出容量暂满，请稍后重试",
            headers={"Retry-After": "5"},
        ) from exc

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。接入现有 HTTP 提交路径后，容量满返回 503、Retry-After: 5；失败提交不返回 202 或一个实际不存在的 job_id。

刚才发生了什么

整个服务暂时容纳不下新导出时，503 给调用者一个明确的暂时失败。只有任务和事件确实提交后才算受理成功。

检查学习结果

• 并发提交不会让受理记录超过配置容量。
• fast与bulk事件分别被对应worker消费。
• 过载时返回503而不是202，恢复后最老等待逐步下降。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

大批量导出占满worker，小型用户导出等待过久。

实现决策

• 入口容量、worker并发、数据库连接和下游速率分别设限。
• 队列隔离有对应的消费者资源，不能只换名称。
• 积压监控同时看数量与最老等待，outbox未发布也计入受理容量。

隔离环境演练

仅暂停本地bulk消费者，提交大小两类任务，观察fast仍处理；随后压到小容量上限确认拒绝。

恢复验收

• 受理总量有可验证上限。
• 关键队列保留处理能力。
• 503请求没有留下已受理但无状态的记录。

遇到问题时查这里

队列分了级，小任务仍一直等待

原因：所有队列仍由同一批已被长任务占满的worker处理。

处理：为关键小任务保留独立消费者和资源预算。

检查count后仍突破容量

原因：多个producer并发读取相同count，检查与INSERT不在同一串行化范围。

处理：同一事务获取约定锁，检查后写job与outbox，所有入口遵守。

自己试一次

将本地受理上限改为5，用两个producer同时提交，核对queued/running总数没有超过5。

提示：先处理或隔离此前活动任务，再做小容量实验。

参考答案：

超过容量的事务抛CapacityExceeded并回滚；只有实际写入状态表和outbox的请求能返回202。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

增加worker不等于无限扩容

完成第 2 步“在受理事务内检查容量”后，再回顾这条规则：系统吞吐还受数据库连接、存储I/O和CPU限制。到达速度长期高于完成速度时，队列持续增长，必须在入口拒绝、等待或降级。

限并发与限速分别处理

完成第 5 步“用数据库观察积压与最老等待”后，再回顾这条规则：concurrency控制同时执行，rate limit控制时间窗口速率；Celery任务rate_limit通常按worker实例生效，不等于全局额度。全局限制应在共享状态或独立入口实现。

队列拆分需要对应消费者容量

完成第 4 步“分别启动两个受限worker”后，再回顾这条规则：给fast和bulk命名不会自动隔离资源；分别启动受限worker，并配置CPU、连接等预算。prefetch降低预取囤积，但不能代替入队容量限制。

固定门闩与隔离级别配合

完成第 2 步“在受理事务内检查容量”后，再回顾这条规则：每次受理先锁住独立 gate 表的 id=1，再在 READ COMMITTED 下计数和写入。所有生产者遵循同一锁顺序和容量值；普通先 count 后 insert 无法防止并发超额。

技术文档

• Celery预取与容量
• Celery任务路由
• MySQL 8.4 锁定读取

---

14 用持久时间槽合并周期执行，并验证跨分钟重投边界

每分钟生成一份商品清单时，重复的调度信号会不会多建任务？用一个可查的时间标签合并同一分钟，再检验跨分钟重复的边界。

从一个问题开始

每分钟生成一份商品清单时，重复的调度信号会不会多建任务？用一个可查的时间标签合并同一分钟，再检验跨分钟重复的边界。

这次的终点

同一时间槽只建一项任务，下一分钟可建新任务；明确记录停机不补跑、跨分钟重投会进入另一槽的现象。

开始前，确认你在这里

• 已完成任务第 13 课，容量门闩、fast/bulk worker 与 publisher 已可用；在 tasks_lab 继续添加文件，MySQL 会话使用 UTC 并连接 shop_lab。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。
• worker、producer、publisher 均在同一个 Linux/WSL2 环境的 tasks_lab 启动；Bash 用 source .venv/bin/activate。环境变量只在当前终端生效，新终端按本课文件加载 .env.tasks，不复制 Windows 虚拟环境。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 用持久时间槽和唯一约束让同一分钟只安排一次导出
• 让 beat 发送周期信号、worker 创建任务、publisher 发送导出消息
• 验证同槽重复不新增，跨分钟重投产生新槽且停机分钟不自动补跑

一步步完成案例

1. 创建时间槽唯一约束

在 UTC 会话执行建表。MySQL 外键立即检查，周期事务不能先写一个引用不存在任务的时间槽。改为先锁定第 13 课的容量门闩，检查该槽是否已有任务；新槽按 job → slot → outbox 顺序写入并一起提交。slot_key 唯一约束继续承担最终校验，异常必须回滚整个创建事务。

文件：schema_schedule.sql

CREATE TABLE IF NOT EXISTS export_schedule_slots (
  slot_key VARCHAR(160) CHARACTER SET ascii COLLATE ascii_bin PRIMARY KEY,
  job_id CHAR(36) CHARACTER SET ascii COLLATE ascii_bin NOT NULL UNIQUE,
  created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  CONSTRAINT export_schedule_job_fk FOREIGN KEY (job_id) REFERENCES export_jobs(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

你应该看到什么

export_schedule_slots 的 slot_key 是主键，job_id 唯一并引用已有任务；表结构可查且使用 ASCII 区分标签。

刚才发生了什么

时间槽是“这一分钟的这项工作”的唯一名字。数据库唯一约束留下已经安排过的证据；外键要求先有任务，才能记录这个槽属于谁。

2. 在同一事务中创建周期任务

新建 schedule_store.py，复用 transaction、lock_admission、admit_export、read_snapshot。同一槽的并发调用在固定门闩上串行检查，因此重复槽即使容量已满也返回 None；新槽再核对容量。立即外键要求先创建 job，然后插入唯一槽和事件。不要捕获任意重复键错误后直接提交，否则可能留下多余 job。这个小练习所有周期共用门闩；生产可按租户设计固定顺序的配额记录。

文件：schedule_store.py

import json
from uuid import uuid4
from state_store import transaction, lock_admission, admit_export, read_snapshot


def schedule_once(slot_key: str, product_ids: list[int]):
    if not isinstance(slot_key,str) or not slot_key.isascii() or not 1 <= len(slot_key) <= 160:
        raise ValueError("slot_key must contain 1..160 ASCII characters")
    job_id = str(uuid4())
    with transaction() as cur:
        lock_admission(cur)
        cur.execute(
            "SELECT job_id FROM export_schedule_slots WHERE slot_key=%s FOR UPDATE",
            (slot_key,),
        )
        if cur.fetchone() is not None:
            return None
        # 同一事务已经持有门闩；再次锁定同一行不会释放已有锁。
        admit_export(cur)
        rows = read_snapshot(cur,product_ids)
        cur.execute(
            """INSERT INTO export_jobs(id,payload,deadline_at)
               VALUES (%s,%s,DATE_ADD(NOW(6), INTERVAL 10 MINUTE))""",
            (job_id,json.dumps(rows,ensure_ascii=False)),
        )
        if cur.rowcount != 1:
            raise RuntimeError("scheduled job insert failed")
        cur.execute(
            "INSERT INTO export_schedule_slots(slot_key,job_id) VALUES (%s,%s)",
            (slot_key,job_id),
        )
        if cur.rowcount != 1:
            raise RuntimeError("schedule slot insert failed")
        cur.execute(
            """INSERT INTO export_outbox(id,job_id,queue_name)
               VALUES (%s,%s,'exports.bulk')""",
            (str(uuid4()),job_id),
        )
        if cur.rowcount != 1:
            raise RuntimeError("scheduled outbox insert failed")
    return job_id

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。同一 slot_key 第一次调用返回 job_id，第二次返回 None；首次的 job、slot 和 outbox 一起存在。

刚才发生了什么

固定门闩让并发检查按顺序完成。写入按 job→slot→outbox 排列是因为 MySQL 当场检查外键，三项仍共同提交或回滚。

3. 注册周期信号并配置beat

新建schedule_tasks.py；将模块加入Celery include并重启worker。把下方app.conf部分放入celery_app.py。此信号使用fast队列，实际导出由outbox进入bulk；继续运行publisher才能把事件变成消息。execution_slot接受显式时间，便于下一个实验固定跨分钟时序；正式tick仍使用开始执行时的now，因此不能宣称保留原beat计划时间。

文件：schedule_tasks.py

from datetime import datetime, timezone
from celery_app import app
from state_store import transaction
from schedule_store import schedule_once


def execution_slot(when: datetime) -> str:
    if when.tzinfo is None:
        raise ValueError("timezone-aware timestamp required")
    return when.astimezone(timezone.utc).strftime("shop-export:%Y%m%dT%H%M")


@app.task(name="products.schedule_tick")
def schedule_tick():
    slot = execution_slot(datetime.now(timezone.utc))
    with transaction() as cur:
        cur.execute("SELECT id FROM products ORDER BY id LIMIT 100")
        rows = cur.fetchall()
    ids = [row["id"] for row in rows]
    return schedule_once(slot,ids) if ids else None

# 以下配置写入 celery_app.py：
# app.conf.timezone = "UTC"
# app.conf.beat_schedule = {
#   "export-each-minute": {
#     "task": "products.schedule_tick", "schedule": 60.0,
#     "options": {"queue": "exports.fast"},
#   }
# }

你应该看到什么

worker 的任务列表出现 products.schedule_tick，配置把周期信号送 fast、实际导出事件送 bulk；publisher 继续把事件发出去。

刚才发生了什么

beat 只按时间发送提醒，worker 收到提醒后才决定当前时间槽并建任务。本例用开始执行的 UTC 分钟，因此延迟执行可能落入另一个槽。

4. 启动一个beat并验证重复信号

Linux/WSL2单独终端执行命令，启动文件只给本次beat使用。不要为了高可用复制同一个本地schedule文件。用schedule_once对同一个测试slot连续调用两次，第一次返回job_id，第二次None。 注释示例中的 [1] 必须替换成 SELECT MIN(id) FROM products 得到的本次真实商品编号；测试槽也换成本次未使用的新名字。

文件：终端 / 当前课时

python -m celery -A celery_app:app beat -l INFO --schedule=./beat-lab.db
# 另一个终端在Python里测试同槽去重：
# from schedule_store import schedule_once
# print(schedule_once("drill:unique-slot-01", [1]))
# print(schedule_once("drill:unique-slot-01", [1]))

你应该看到什么

只运行一个 beat 进程时可见周期信号；对新的测试槽连续两次调用分别返回编号和 None。

刚才发生了什么

调度发送和执行去重各有责任：一个 beat 减少重复信号，持久槽仍负责挡住同槽重复。更换进程也不会让数据库忘记已经安排的槽。

5. 证明同槽去重与跨分钟重投的区别

本实验直接调用同一slot生成逻辑，并注入两个明确的UTC时间，不必等待真实分钟切换。相同delivery_label表示模拟同一信号；两次不同分钟仍会创建两个job，结果是当前模型的边界，不应标成原消息exactly-once。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python verify_schedule_slots.py，再与下方预期对照。

文件：verify_schedule_slots.py

from datetime import datetime, timezone
from uuid import uuid4
from state_store import transaction
from schedule_store import schedule_once
from schedule_tasks import execution_slot

with transaction() as cur:
    cur.execute("SELECT MIN(id) AS id FROM products")
    product_id = cur.fetchone()["id"]
assert product_id is not None
prefix = f"drill:{uuid4()}:"
delivery_label = "same-redelivered-signal"
t1 = datetime(2026,1,1,10,0,59,tzinfo=timezone.utc)
t2 = datetime(2026,1,1,10,1,1,tzinfo=timezone.utc)
first = schedule_once(prefix+execution_slot(t1),[product_id])
same_slot = schedule_once(prefix+execution_slot(t1),[product_id])
later_slot = schedule_once(prefix+execution_slot(t2),[product_id])
assert first is not None and same_slot is None
assert later_slot is not None and later_slot != first
print(delivery_label,{"first":first,"same_slot":same_slot,"later_slot":later_slot})
# 创建两条本地 job 与 outbox，继续由正常 publisher/worker 处理。

你应该看到什么

脚本打印 first 编号、same_slot=None、不同的 later_slot 编号，所有断言通过；数据库增加两项任务及对应事件。

刚才发生了什么

同一信号在两个执行分钟得到不同槽名，所以本模型只保证同槽合并。若要按原计划时间去重，必须随消息保留计划时间或持久调度游标。 每次生成新的drill前缀避免与真实调度或此前实验混用。

6. 验证停机与时区行为

停止本地beat两分钟再启动，查询时间槽表。本例不会自动补发停机期间每一分钟的导出；明确记录这一行为。若需要补跑，先设计持久due_at游标和补跑容量，再修改任务。

文件：终端 / 当前课时

SELECT slot_key,job_id,created_at
FROM export_schedule_slots ORDER BY created_at DESC LIMIT 10;

你应该看到什么

停 beat 两分钟再恢复后，槽表只显示实际执行时创建的分钟，缺少的每分钟不会自动补齐。

刚才发生了什么

时间字段统一 UTC 避免本地时区改变槽名。补跑需要单独保存“下一次应执行的时间”并限制补跑数量，本节观察到的行为没有这项能力。

检查学习结果

• 同一slot_key最多对应一个job_id与初始outbox事件。
• beat只发送信号，文件由worker和publisher链路完成。
• 时区固定UTC，停机遗漏的分钟不会被伪装成已执行。
• 同一模拟信号跨分钟执行会生成两个槽，明确记录这与同槽去重的区别。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

每分钟为商品生成一份小快照，调度进程偶尔重启。

实现决策

• 数据库唯一键合并同一执行分钟的任务；本课不承诺原始beat信号跨分钟重投只创建一次。
• 定义时区、时间槽和停机补跑语义，不能依靠默认本地时间。
• 慢任务的重叠策略和补跑容量单独控制。
• 需要固定计划时间和补跑时，升级为持久next_due_at游标与唯一scheduled_at槽，受理额度不足时不推进游标。

隔离环境演练

依次验证同槽重复调用、同一模拟信号跨分钟调用、beat停机两分钟三种情形；当前模型分别得到合并、新槽、无自动补跑。

恢复验收

• 重复调度不能创建两份同槽业务导出。
• 没有把beat当任务执行器或持久业务状态真源。
• 停机后的补跑规则可明确说明。
• 不把执行分钟合并误写成原始信号exactly-once；跨分钟行为有实验证据。

遇到问题时查这里

启动两个beat后定时任务数量翻倍

原因：多个调度器同时发送同一计划。

处理：常态运行单beat，数据库slot唯一约束兜住重复业务创建。

以为周期调度天然不重叠

原因：上一分钟任务未完成，下一分钟仍会生成新任务。

处理：按业务增加导出类型并发限制或跳过策略，保留任务执行租约。

自己试一次

用两个producer并发调用同一全新slot_key，确认只有一个返回job_id。

提示：所有周期创建先锁定同一门闩，再检查槽；没有门闩的普通 SELECT 后 INSERT 仍存在竞争。

参考答案：

第一方在同一事务创建 job、slot 和 outbox 并提交；第二方取得门闩后读到现有槽，返回 None。唯一键继续兜底，立即外键要求先写 job；任何后续写入异常都回滚整个创建事务。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

beat负责发送，worker负责执行

完成第 3 步“注册周期信号并配置beat”后，再回顾这条规则：本课按worker实际开始处理时的UTC分钟生成slot_key，数据库唯一约束保证同一槽只创建一个job。这是按执行分钟合并，不是按原始beat信号去重：同一消息如果跨分钟重投，会落到不同槽并可能新建任务。

时间槽固定一个业务周期

完成第 1 步“创建时间槽唯一约束”后，再回顾这条规则：本地例子用worker处理信号时的UTC分钟作为slot_key，同一分钟最多创建一份快照。它不补齐停机错过的分钟；需要严格补跑时，应由持久调度表保存due_at与补跑规则。

不重叠执行还需要租约

完成第 2 步“在同一事务中创建周期任务”后，再回顾这条规则：同一时间槽不重复创建，并不阻止不同时间槽的大导出重叠。复用任务领取租约，并按业务决定跳过、合并或限制同一导出类型的并发。

需要计划时间语义时使用持久调度游标

完成第 5 步“证明同槽去重与跨分钟重投的区别”后，再回顾这条规则：若业务要求每个计划分钟最多一次且停机后补跑，应在数据库保存schedule_id、next_due_at和唯一(schedule_id,scheduled_at)槽。调度事务锁定配置、按有限批次推进游标并写job/outbox；worker消费持久scheduled_at，不能再用执行时now猜测原计划。本课保留较小的执行分钟模型，下面用实际两次调用证明边界。

立即外键决定写入顺序

完成第 2 步“在同一事务中创建周期任务”后，再回顾这条规则：MySQL 在执行语句时检查外键。先持有固定门闩，再检查时间槽；创建时按 job、slot、outbox 的顺序写入，同一事务提交。事务结束前不发送消息。

技术文档

• Celery周期任务
• MySQL 8.4 外键约束
• Python时区时间

---

15 实现优雅退出、取消与失败隔离恢复

任务可能被取消、执行者可能退出，也可能消息已发出却无人领取。把这些未完成记录重新检查，给可恢复的任务一次有上限的机会。

从一个问题开始

任务可能被取消、执行者可能退出，也可能消息已发出却无人领取。把这些未完成记录重新检查，给可恢复的任务一次有上限的机会。

这次的终点

取消任务不能发布成功；过期和未领取任务生成恢复事件且遵守退避，耗尽任务留有失败记录，修复后可另建任务重放。

开始前，确认你在这里

• 已完成任务第 11–14 课，state_store.py、publisher.py 与 fast/bulk 路由齐备；在 shop_lab 只操作新建实验 job_id，先确认可正常生成一个结果文件。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。
• worker、producer、publisher 均在同一个 Linux/WSL2 环境的 tasks_lab 启动；Bash 用 source .venv/bin/activate。环境变量只在当前终端生效，新终端按本课文件加载 .env.tasks，不复制 Windows 虚拟环境。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 在工作边界和完成写回时核对取消、期限与当前令牌
• 验证租约过期和已派发未领取任务会恢复，重复消息不能越过退避
• 让次数耗尽任务保留失败证据，修复后建立带 replay_of 的新任务

一步步完成案例

1. 增加取消标记与隔离表

在 UTC 会话建立业务失败隔离表并单独执行一次迁移。取消只修改经过归属校验的活跃记录，终态不回退；重复取消可能没有字段变化，不能仅凭受影响行数断言任务不存在。文件操作可能继续片刻，但 finish 必须检查 cancel_requested 和 deadline。

文件：schema_recovery.sql

CREATE TABLE IF NOT EXISTS export_dead_letters (
  job_id CHAR(36) CHARACTER SET ascii COLLATE ascii_bin PRIMARY KEY,
  reason VARCHAR(64) NOT NULL,
  quarantined_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  CONSTRAINT export_dead_letter_job_fk FOREIGN KEY (job_id) REFERENCES export_jobs(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 以下迁移只执行一次；已有 replay_of 时先核对 SHOW CREATE TABLE。
ALTER TABLE export_jobs
  ADD COLUMN replay_of CHAR(36) CHARACTER SET ascii COLLATE ascii_bin,
  ADD CONSTRAINT export_jobs_replay_fk FOREIGN KEY (replay_of) REFERENCES export_jobs(id);

-- 取消由应用在短事务中执行参数化 UPDATE，随后读取结果。
-- UPDATE export_jobs SET cancel_requested=TRUE,updated_at=NOW(6)
-- WHERE id=%s AND status IN ('queued','running');
-- 重复取消可能 rowcount=0；再 SELECT 核对记录、归属及状态，不误报不存在。

你应该看到什么

export_dead_letters 表存在，export_jobs 增加 replay_of 外键；取消请求只更新 queued/running 的指定记录。

刚才发生了什么

取消是持久意愿，执行者要读取后配合停止。失败隔离表保留为什么不再自动执行；replay_of 把修复后的新任务关联到原记录。 取消接口需事务内读取最终状态。

2. 在处理边界检查权属

把can_continue追加到state_store.py，在生成每个数据块前调用，返回False就停止后续工作。小文件实验可在write_text前检查；完成CAS仍不可省略。有效租约也必须处于业务deadline以内。

文件：state_store.py

def can_continue(job_id: str, token: str) -> bool:
    with transaction() as cur:
        cur.execute("""
            SELECT id FROM export_jobs
            WHERE id=%s AND lease_token=%s AND status='running'
              AND lease_until>NOW(6) AND deadline_at>NOW(6)
              AND cancel_requested=FALSE
        """, (job_id,token))
        row = cur.fetchone()
    return row is not None

# export_tasks.py 生成文件前先导入并调用：
# from state_store import can_continue
# if not can_continue(job_id,token):
#     return {"job_id":job_id,"published":False}

你应该看到什么

本步先保存定义，不会自动产生运行输出；调用后的观察在后续验证步骤完成。生成文件前 can_continue 对有效持有者返回 True，取消、超时或令牌改变后返回 False；finish 仍保留最终条件检查。

刚才发生了什么

工作开始时检查一次不能保护之后整段执行。分块边界检查让任务尽快响应变化，最后的 CAS 则防止检查后到写回前再次失去权属。

3. 在publisher确认时记录派发时间

替换 publisher.py 的 publish_one，保留 claim_event，并确认第 13 课已让 SELECT 返回 queue_name。确认事务先锁 job，再更新 outbox，与扫描器的锁顺序一致；CAS 匹配当前 token、未确认状态和有效领取租约后才更新 last_dispatched_at。事件确认与派发时间同事务提交。旧 publisher 不能推迟恢复判断；派发成功仍不能保证 broker 消息永不丢失。发生死锁或连接异常应回滚、告警并进行有预算的外层重试，不能把未知结果标成成功。

文件：publisher.py

def publish_one():
    event = claim_event()
    if event is None:
        return False
    app.send_task(
        "products.export",args=[event["job_id"]],
        task_id=event["id"],queue=event["queue_name"],retry=False,
    )
    with transaction() as cur:
        # 与 recovery 采用相同顺序：先 job，后 outbox。
        cur.execute("SELECT id FROM export_jobs WHERE id=%s FOR UPDATE",(event["job_id"],))
        if cur.fetchone() is None:
            raise RuntimeError("outbox refers to a missing job")
        cur.execute("""
            UPDATE export_outbox SET sent_at=NOW(6),claim_until=NULL
            WHERE id=%s AND job_id=%s AND claim_token=%s AND sent_at IS NULL
              AND claim_until>NOW(6)
        """,(event["id"],event["job_id"],event["claim_token"]))
        confirmed = cur.rowcount == 1
        if confirmed:
            cur.execute(
                "UPDATE export_jobs SET last_dispatched_at=NOW(6) WHERE id=%s",
                (event["job_id"],),
            )
    return confirmed

你应该看到什么

发送成功且当前事件确认成功时，sent_at 与任务 last_dispatched_at 一起写入；旧 token 确认失败不会推迟任务派发时间。

刚才发生了什么

发送时间只能证明发布器完成了派发，不能证明 worker 已领取。先锁 job 再锁 outbox 的固定顺序，也让发布器和扫描器更少互相卡住。 网络发送仍在事务外。

4. 扫描租约与未领取任务，原子生成恢复事件

在 tasks_lab 新建 recovery.py，保留本课固定 job→outbox 的加锁顺序与每批 50 条上限。先保留正常任务作基线，运行 python recovery.py 后用本课第 5、6 步新建的专用 job_id 验证不同恢复路径。长期运行扫描器时应空批等待、异常告警、停机时停止取新批次；事务内不发消息、不写文件、不等待网络。

文件：recovery.py

from uuid import uuid4
from state_store import transaction


def recover_batch():
    with transaction() as cur:
        cur.execute("""
            SELECT j.*,j.deadline_at<=NOW(6) AS expired
            FROM export_jobs AS j
            WHERE (j.status='running' AND j.lease_until<NOW(6))
               OR (j.status='queued' AND
                   (j.cancel_requested=TRUE OR j.deadline_at<=NOW(6) OR j.attempts>=3))
               OR (j.status='queued' AND j.next_attempt_at<=NOW(6)
                   AND j.cancel_requested=FALSE AND j.deadline_at>NOW(6) AND j.attempts<3
                   AND NOT EXISTS (
                     SELECT 1 FROM export_outbox AS pending
                     WHERE pending.job_id=j.id AND pending.sent_at IS NULL)
                   AND GREATEST(j.updated_at,j.next_attempt_at,
                       COALESCE(j.last_dispatched_at,
                         (SELECT MAX(o.sent_at) FROM export_outbox AS o WHERE o.job_id=j.id),
                         j.created_at)) < DATE_SUB(NOW(6), INTERVAL 60 SECOND))
            ORDER BY j.updated_at,j.id LIMIT 50 FOR UPDATE SKIP LOCKED
        """)
        rows = cur.fetchall()
        for row in rows:
            if row["cancel_requested"]:
                status,reason,delay = "cancelled","user_cancelled",0
            elif row["expired"] or row["attempts"] >= 3:
                status = "failed"
                reason = "deadline_exceeded" if row["expired"] else "attempts_exhausted"
                delay = 0
            elif row["status"] == "running":
                status,reason,delay = "queued","lease_expired",10
            else:
                status,reason,delay = "queued","dispatch_overdue",0
            cur.execute("""
                UPDATE export_jobs SET status=%s,error_code=%s,
                  lease_token=NULL,lease_until=NULL,
                  next_attempt_at=TIMESTAMPADD(SECOND,%s,NOW(6)),updated_at=NOW(6)
                WHERE id=%s AND status=%s AND lease_token <=> %s
            """,(status,reason,delay,row["id"],row["status"],row["lease_token"]))
            if cur.rowcount != 1:
                raise RuntimeError("locked recovery row did not match")
            cur.execute("SELECT next_attempt_at FROM export_jobs WHERE id=%s",(row["id"],))
            updated = cur.fetchone()
            if updated is None:
                raise RuntimeError("recovered job disappeared")
            if status == "failed":
                cur.execute(
                    "INSERT INTO export_dead_letters(job_id,reason) VALUES (%s,%s)",
                    (row["id"],reason),
                )
                if cur.rowcount != 1:
                    raise RuntimeError("dead letter was not recorded")
            if status == "queued":
                cur.execute("""
                    INSERT INTO export_outbox(id,job_id,queue_name,available_at)
                    VALUES (%s,%s,'exports.bulk',%s)
                """,(str(uuid4()),row["id"],updated["next_attempt_at"]))
                if cur.rowcount != 1:
                    raise RuntimeError("recovery event was not recorded")
    return len(rows)


if __name__ == "__main__":
    print("recovered",recover_batch())

你应该看到什么

python recovery.py 打印本次处理数量；过期 running 变 queued 并设置 10 秒退避，有 pending 的 queued 不重复堆事件，耗尽任务进入 failed。

刚才发生了什么

扫描器根据持久状态找遗漏，把状态变化和恢复事件一起提交。未领取的恢复不增加执行次数，真正 claim 才增加 attempts，避免把网络交付次数混作业务执行次数。 每批最多 50 条，在 READ COMMITTED 短事务内 SELECT ... FOR UPDATE SKIP LOCKED 领取 job，再更新状态并写 outbox。 对 running 租约过期的任务，恢复先设置 10 秒 next_attempt_at，再把同一时间写入事件 available_at；两道检查一起拦住提前重试。queued 超过最近派发 60 秒仍未领取且没有 pending 事件时补发；最多 3 次实际领取，业务总截止为 10 分钟。这些是本机实验预算，实际恢复耗时还要加扫描间隔和排队时间。

5. 验证重复消息不能跳过持久退避

先暂停本机publisher与consumer，防止实验记录被正常流程抢先处理。只针对新建实验job设置next_attempt_at为未来10秒；两次直接claim模拟已有重复消息。演练需要10秒等待，只在本地执行；任务后续通过正常worker输出或恢复扫描到终态。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python verify_retry_gate.py，再与下方预期对照。

文件：verify_retry_gate.py

import time
from state_store import transaction,submit_snapshot,claim

with transaction() as cur:
    cur.execute("SELECT MIN(id) AS id FROM products")
    product_id = cur.fetchone()["id"]
assert product_id is not None
job_id = submit_snapshot([product_id])
with transaction() as cur:
    cur.execute(
        "UPDATE export_jobs SET next_attempt_at=DATE_ADD(NOW(6), INTERVAL 10 SECOND) WHERE id=%s",
        (job_id,),
    )
assert claim(job_id) is None
assert claim(job_id) is None
with transaction() as cur:
    cur.execute("SELECT attempts FROM export_jobs WHERE id=%s",(job_id,))
    assert cur.fetchone()["attempts"] == 0
time.sleep(10.1)
job = claim(job_id)
assert job is not None and job["attempts"] == 1
print("persistent retry gate verified",job_id)
# 此处持有真实新租约，等待到期再让 recovery 重排，不伪造成功文件。

你应该看到什么

打印 persistent retry gate verified；未来 10 秒内两次 claim 都为 None、attempts=0，到期后成功领取并变为 1。

刚才发生了什么

next_attempt_at 写在数据库中，所有重复消息都必须经过它。只把延迟放在一条消息上，挡不住另一条已在队列里的重复消息提前执行。 两次均为None，attempts保持0。到期后claim成功且attempts变1。

6. 模拟派发后消息丢失，不删除broker数据

暂停本机publisher和consumer后执行本实验，避免正常进程抢先处理。使用新建job的outbox行模拟发送已确认，但不真的发送Celery消息；只修改该job时间。扫描后必须出现新的pending outbox，第二次扫描不能再给它新增事件。恢复publisher和consumer后应正常执行。这覆盖排队消息丢失，也覆盖worker在claim前数据库连接失败的相同持久状态。 将本步代码按标注文件位置保存或合并，完成上述前置后在当前项目终端执行 python verify_queued_recovery.py，再与下方预期对照。

文件：verify_queued_recovery.py

from state_store import transaction,submit_snapshot
from recovery import recover_batch

with transaction() as cur:
    cur.execute("SELECT MIN(id) AS id FROM products")
    product_id = cur.fetchone()["id"]
assert product_id is not None
job_id = submit_snapshot([product_id])
with transaction() as cur:
    # 本实验先暂停 publisher / worker，仅修改本次新建 job。
    cur.execute("SELECT id FROM export_jobs WHERE id=%s FOR UPDATE",(job_id,))
    assert cur.fetchone() is not None
    cur.execute("""
        UPDATE export_outbox SET sent_at=DATE_SUB(NOW(6), INTERVAL 61 SECOND),claim_until=NULL
        WHERE job_id=%s
    """,(job_id,))
    cur.execute("""
        UPDATE export_jobs SET last_dispatched_at=DATE_SUB(NOW(6), INTERVAL 61 SECOND),
          updated_at=DATE_SUB(NOW(6), INTERVAL 61 SECOND),
          next_attempt_at=DATE_SUB(NOW(6), INTERVAL 61 SECOND)
        WHERE id=%s
    """,(job_id,))
recover_batch()
with transaction() as cur:
    cur.execute(
        "SELECT COUNT(*) AS n FROM export_outbox WHERE job_id=%s AND sent_at IS NULL",
        (job_id,),
    )
    assert cur.fetchone()["n"] == 1
recover_batch()
with transaction() as cur:
    cur.execute(
        "SELECT COUNT(*) AS n FROM export_outbox WHERE job_id=%s AND sent_at IS NULL",
        (job_id,),
    )
    assert cur.fetchone()["n"] == 1
    cur.execute("SELECT attempts FROM export_jobs WHERE id=%s",(job_id,))
    assert cur.fetchone()["attempts"] == 0
print("queued delivery recovery verified",job_id)
# 恢复正常 publisher / consumer，再核对该 job 终态与结果文件。

你应该看到什么

连续两次扫描后该 job 的未发送事件数量都为 1，attempts 仍为 0，打印 queued delivery recovery verified。

刚才发生了什么

已确认派发仍可能丢消息，或 worker 在领取前连接数据库失败。超过未领取预算后补一条通知能恢复这类遗漏，检查已有 pending 则避免每次扫描都新增。

7. 先正常停机，再验证租约恢复

先对本次专用 worker 正常发送 TERM 或首次 Ctrl+C，观察 warm shutdown。故障实验仅终止已核对的执行进程，不按模糊名称批量终止。30 秒租约过期后运行 recovery.py；它会把新的恢复事件推迟 10 秒可领取，因此先按本次 job_id 查询 next_attempt_at 和 outbox.available_at，至少等到该时刻，再运行 publisher.py。重启相应 worker 并按本次 job_id 轮询数据库，核对最后的 succeeded/result_ref；如果还有更早事件，单次发布可能先处理其他任务，要按 ID 确认本次派发。

文件：终端 / 当前课时

python -m celery -A celery_app:app inspect active
python -m celery -A celery_app:app inspect reserved
# 正常关闭：仅对已核对的本次 worker PID 发送 TERM，或首次 Ctrl+C。
# 故障实验先等该 job 的 30 秒租约过期，再进行恢复：
python recovery.py
# 本课过期 running 路径新增 10 秒退避；核对本 job 的 available_at。
sleep 11
python publisher.py
# 恢复相应队列的 worker 后，按本次 job_id 核对状态与结果文件。

你应该看到什么

正常关闭可见 warm shutdown；过期任务扫描后先回 queued 并带未来 10 秒的 next_attempt_at。过早发布返回 False 属于未到期；到期发送并恢复 worker 后，本次 job 能再次领取，最终状态和真实文件可核对。

刚才发生了什么

恢复扫描只把任务重新排入可尝试流程，不会让它立即执行。持久退避仍由领取函数与 outbox 同时检查；给出的 sleep 只用于这个本机 10 秒实验，实际是否就绪以数据库的 available_at 为准。正常关闭与故障租约恢复经过不同路径，两者都需要核对最终业务状态。

8. 修复后创建新任务重放

新建replay_job.py，在命令行传一个明确的failed任务ID。事务中保留payload、记录replay_of、建立新outbox；原dead letter不删除。自动批量重放前必须有故障已修复的证据，本模块只逐条本地重放。

文件：replay_job.py

import json
import sys
from uuid import UUID, uuid4
from state_store import transaction, admit_export, decode_payload


def replay(old_id: str):
    old_id = str(UUID(old_id))
    new_id = str(uuid4())
    with transaction() as cur:
        admit_export(cur)
        cur.execute(
            "SELECT id,payload FROM export_jobs WHERE id=%s AND status='failed' FOR UPDATE",
            (old_id,),
        )
        old = cur.fetchone()
        if old is None:
            raise ValueError("failed source job not found")
        payload = decode_payload(old["payload"])
        cur.execute("""
            INSERT INTO export_jobs(id,payload,replay_of,deadline_at)
            VALUES (%s,%s,%s,DATE_ADD(NOW(6), INTERVAL 10 MINUTE))
        """,(new_id,json.dumps(payload,ensure_ascii=False),old_id))
        if cur.rowcount != 1:
            raise RuntimeError("replay job was not created")
        cur.execute("""
            INSERT INTO export_outbox(id,job_id,queue_name)
            VALUES (%s,%s,'exports.bulk')
        """,(str(uuid4()),new_id))
        if cur.rowcount != 1:
            raise RuntimeError("replay event was not created")
    return new_id


if __name__ == "__main__":
    print(replay(sys.argv[1]))

你应该看到什么

对一个确切 failed ID 运行重放脚本返回新 ID；新行 replay_of 指向旧 ID、创建新 outbox，原失败与死信仍保留。

刚才发生了什么

修复后重放是新的业务尝试，不能擦掉原来的失败证据或清零旧任务次数。沿用原快照才能核对修复前后处理的是同一份输入。

检查学习结果

• 取消任务不能被finish发布为成功，最终由执行者或扫描器转cancelled。
• 过期租约恢复产生新事件，attempts不归零，最多3次领取。
• 耗尽任务进入failed和dead letter，重放使用新job_id并记录replay_of。
• 已sent但未领取的queued任务能产生新的恢复事件；已有pending事件时不会每轮扫描继续堆积。
• next_attempt_at到期前重复消息无法增加attempts，派发次数与实际执行次数分开。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

发布新版本需要停worker，用户同时请求取消导出，部分任务已多次崩溃。

实现决策

• 取消、正常退出、强制终止分别定义结果和恢复方式。
• 扫描器只处理过期或明确终止条件，使用短事务与有限批次。
• 故障修复后以新任务重放，保留旧失败状态和审计关联。
• queued久未领取和running租约到期分别恢复；派发确认不作为完成证据。
• 持久next_attempt_at约束所有消息，包括已有重复投递；broker倒计时不是唯一退避控制。
• publisher与恢复扫描必须由持续监督运行的进程定期执行，单次脚本不是生产调度器。

隔离环境演练

依次验证过期租约、未来next_attempt_at收到重复消息、已sent却queued、已有pending事件抑制重复恢复、3次耗尽与单条重放；每次仅操作记录的本地job_id。

恢复验收

• 有效租约不会被扫描器抢占。
• 尝试耗尽不会形成无限重投循环。
• 重放有来源记录且不会抹掉原始失败证据。
• 派发已确认但消息未领取的任务可恢复，不只等deadline后失败。
• 退避时间未到时重复消息不消耗执行attempts。
• 暂时无消费者时重新派发有阈值和deadline限制，不无限快速重投。

遇到问题时查这里

revoke后立即在页面显示已取消

原因：revoke或发送取消信号不等于所有执行者与外部操作已停止。

处理：先显示取消申请，待持久状态确认后显示cancelled。

恢复脚本把所有running直接改queued

原因：仍有有效租约的执行者会与新worker竞争。

处理：只恢复过期租约，旧执行者发布必须受token、时间和状态CAS约束。

outbox已sent但任务一直queued

原因：消息可能丢失，或worker在claim之前因数据库错误退出；只恢复running无法覆盖。

处理：按持久派发时间扫描久未领取且无pending事件的queued任务，原子重建发送意图；仍用claim防重复执行。

自己试一次

连续让同一隔离任务领取后租约过期，验证第3次后进入failed，不产生第4次领取。

提示：只操作本次新建任务，并通过正常claim增加attempts。

参考答案：

前两次过期产生恢复事件；第三次过期被扫描器隔离，attempts保持3，后续claim返回None；原错误和重放链可查询。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

取消需要执行者配合

完成第 2 步“在处理边界检查权属”后，再回顾这条规则：取消标志表示用户请求停止，不能证明线程、子进程或外部写入已经停止。执行者在分块边界检查权属与取消，发布结果前再次CAS；不要把terminate当普通业务取消。

正常退出先停止取新任务

完成第 7 步“先正常停机，再验证租约恢复”后，再回顾这条规则：Celery worker收到TERM进行warm shutdown，等待正在运行工作；强制终止不能保证finally执行。退出宽限期要覆盖有界单次任务，恢复依赖状态租约和未确认消息机制。

死信是可核对的业务隔离

完成第 8 步“修复后创建新任务重放”后，再回顾这条规则：Redis broker没有自动替本模块建立业务死信流程。本课用export_dead_letters保存尝试耗尽或过期任务，原任务保持failed；修复后创建新job重放，保留原始失败证据。

派发成功与任务领取是不同事实

完成第 6 步“模拟派发后消息丢失，不删除broker数据”后，再回顾这条规则：publisher收到发送成功后，消息仍可能在broker故障中丢失；worker也可能在claim之前因数据库连接错误退出。两者都可能留下queued和已sent事件。恢复扫描还要查久未领取、没有未发送事件的queued任务，原子新建outbox。重复派发允许存在，数据库claim决定谁能执行。

技术文档

• Celery worker关闭与revoke
• MySQL 8.4 锁定读取
• Celery任务确认

---

16 完成故障注入与可观测性验收

链路正常跑过一次还不足以说明能恢复。现在按任务编号追踪提交、派发、领取和结果，用一次只改一处的实验验证承诺。

从一个问题开始

链路正常跑过一次还不足以说明能恢复。现在按任务编号追踪提交、派发、领取和结果，用一次只改一处的实验验证承诺。

这次的终点

每个已运行故障都有实际状态、日志、结果文件和恢复耗时证据，未运行项目保留 not_run。

开始前，确认你在这里

• 已完成任务第 01–15 课，正常链路能得到 succeeded 及可读文件；在 tasks_lab 保留 worker、publisher、recovery，MySQL 只读观测会话确认 UTC 与 shop_lab。
• 本课文件名相对 tasks_lab；第 01 课创建后，每个新终端都先进入 tasks_lab 并激活它自己的 .venv，不使用 shop-service 根目录的 .venv。用 python -c "import sys; print(sys.executable)" 核对当前解释器。
• worker、producer、publisher 均在同一个 Linux/WSL2 环境的 tasks_lab 启动；Bash 用 source .venv/bin/activate。环境变量只在当前终端生效，新终端按本课文件加载 .env.tasks，不复制 Windows 虚拟环境。

每次只做一个步骤，先核对结果，再读原因。结果不一致时，先排查当前步骤，再继续往下。

这节课要完成什么

• 按业务编号、消息编号、令牌和阶段日志追踪一次导出
• 核对积压年龄、状态终点与真实结果文件内容
• 逐项注入隔离故障，保存实测恢复耗时并保留未运行项目的 not_run

一步步完成案例

1. 给worker添加结构化阶段日志

在worker领取后和完成后记录下面格式，使用真实变量。错误日志保留异常类型与关联ID，商品描述快照不进入日志。要记录Celery消息ID可将任务改为bind=True，并使用self.request.id；同时保持对外任务名products.export。

文件：task_logging.py

import json
import logging

logger = logging.getLogger("shop.exports")

def log_phase(job_id, phase, message_id=None, lease_token=None, attempt=None, **fields):
    logger.info(json.dumps({
        "event": "export_phase", "job_id": str(job_id),
        "phase": phase, "message_id": message_id,
        "lease_token": str(lease_token) if lease_token else None,
        "attempt": attempt, **fields,
    }, ensure_ascii=False))

# log_phase(job_id, "claimed", self.request.id, token, job["attempts"])
# log_phase(job_id, "published", self.request.id, token, job["attempts"],
#           success=published)

你应该看到什么

领取与完成后出现包含 job_id、phase、attempt、lease_token 的 JSON 日志；相同 job 可关联多条 message_id。

刚才发生了什么

业务任务编号把全过程连起来，消息编号区分重复交付，令牌区分执行权属。阶段日志帮助判断任务卡在哪一步，而不是只得到一句“失败了”。

2. 查询积压、租约和失败

在 UTC 的只读观测会话查询，记录总量、最老等待与对应时段，不把 COUNT=0 时 MIN 的 NULL 年龄写成 0 秒。queued 没有 pending 事件不一定已超恢复阈值，要一起看派发时间、next_attempt_at 和业务 deadline。

文件：monitor_exports.sql

SELECT status,COUNT(*) AS total
FROM export_jobs GROUP BY status;

SELECT COUNT(*) AS unpublished,
       TIMESTAMPDIFF(MICROSECOND,MIN(created_at),NOW(6))/1000000 AS oldest_unpublished_seconds
FROM export_outbox WHERE sent_at IS NULL;

SELECT COUNT(*) AS expired_leases
FROM export_jobs WHERE status='running' AND lease_until<NOW(6);

SELECT reason,COUNT(*) AS failed
FROM export_dead_letters GROUP BY reason;

SELECT COUNT(*) AS queued_without_pending,
       TIMESTAMPDIFF(MICROSECOND,MIN(COALESCE(j.last_dispatched_at,j.created_at)),NOW(6))/1000000
         AS oldest_dispatch_age
FROM export_jobs AS j
WHERE j.status='queued' AND NOT EXISTS (
  SELECT 1 FROM export_outbox AS o WHERE o.job_id=j.id AND o.sent_at IS NULL
);

SELECT id,status,attempts,next_attempt_at,last_dispatched_at
FROM export_jobs WHERE status IN ('queued','running') ORDER BY created_at LIMIT 20;

你应该看到什么

查询得到各状态数量、未发事件最老年龄、过期租约和失败原因；无对应行时年龄为 NULL，应原样记录。

刚才发生了什么

总量、等待年龄和失败原因分别回答“多少、多久、为什么”。queued 没有 pending 不自动代表丢失，要和最近派发时间及恢复阈值一起判断。 TIMESTAMPDIFF 以微秒计算再换算秒。指标标签保持低基数；具体 job_id 放日志。

3. 以任务ID检查终态和文件

创建 verify_terminal.py 并传入本次真实 job_id。PyMySQL 返回 JSON 字符串，先 decode_payload 再与文件的 JSON 值比较，不能比较序列化文本的空格或键顺序。文件必须位于本次 artifacts 目录；生产共享目录软链应按部署路径解析。数据库读取结束后立即释放连接，再做文件读取。

文件：verify_terminal.py

import json
import sys
from pathlib import Path
from uuid import UUID
from state_store import transaction,decode_payload

job_id = str(UUID(sys.argv[1]))
with transaction() as cur:
    cur.execute("SELECT * FROM export_jobs WHERE id=%s",(job_id,))
    job = cur.fetchone()
assert job is not None, "job missing"
print("state",job["status"],"attempts",job["attempts"])
if job["status"] == "succeeded":
    root = Path("artifacts").resolve()
    target = Path(job["result_ref"]).resolve()
    assert target.is_relative_to(root), "unexpected artifact location"
    assert json.loads(target.read_text(encoding="utf-8")) == decode_payload(job["payload"])
    assert not job["cancel_requested"]
    print("artifact verified")
elif job["status"] in {"failed","cancelled"}:
    assert job["result_ref"] is None
    print("terminal without published artifact")

你应该看到什么

成功任务打印 state succeeded 与 artifact verified；失败/取消打印 terminal without published artifact；文件不匹配会触发断言。

刚才发生了什么

最终验收同时读数据库和文件，防止状态写成功而文件缺失。比较 JSON 值而非文本排版，保证检查的是商品内容。

4. 按矩阵逐个注入故障

每一行使用新建job_id，先记基线，再暂停或修改本次隔离组件，最后恢复并运行publisher/recovery。不要同时注入多种故障，否则无法确定恢复由哪个机制保证。下面是待执行清单，不代表已经跑过。

文件：failure-matrix.md

| 故障点 | 操作 | 应核对的结果 |
|---|---|---|
| DB事务提交前 | 本次提交主动抛异常 | job与outbox同时回滚 |
| broker不可用 | 暂停本地Redis后提交 | job与outbox保留，sent_at为空 |
| 发布后未标sent | 跳过一次确认写回 | 可重复发布，业务结果仍受CAS保护 |
| worker持租约退出 | 终止本次隔离执行进程 | 到期恢复，attempts增加 |
| 旧token晚到 | 新领取后用旧token完成 | finish=False，不改result_ref |
| 取消 | 设置本次cancel_requested | 最终cancelled，不发布成功结果 |
| 次数耗尽 | 3次领取均失败/过期 | failed及dead letter，无第4次领取 |
| 结果缓存过期 | 缩短本地result_expires | MySQL状态仍可查询 |
| 队列满 | 临时小容量并发提交 | 503，无虚假202 |
| 同槽周期重复 | 相同slot并发调度 | 只创建一个周期job |
| 周期跨分钟重投 | 同一模拟信号用两个执行分钟 | 当前模型创建两个槽，不宣称原信号exactly-once |
| 已派发但未领取 | 专用job模拟sent且queued、时间超过阈值 | 扫描生成新outbox，恢复后完成 |
| claim之前失败 | worker进入claim前抛连接错误 | queued路径能重新派发 |
| 退避期间重复消息 | next_attempt_at未来时直接重复claim | 均返回None，attempts不增加 |
| pending事件仍在 | 同一个queued重复运行扫描器 | 不为已有pending事件继续新增恢复事件 |

你应该看到什么

每项故障使用独立 job_id，记录矩阵对应现象；完成一项恢复后再做下一项，未执行的行不写通过。

刚才发生了什么

故障注入是有意打断某一步，再观察持久记录如何收敛。一次只注入一处，才知道是哪种保护机制让任务恢复或可靠失败。

5. 保存观测证据并核对恢复预算

为每次演练填写实际值：操作时间、job_id、消息ID、租约、状态转换、输出摘要、恢复耗时。只有实际运行后才能标pass。设置本地验收阈值，例如过期租约路径在60秒内重新进入处理、queued丢消息路径在90秒内重新进入处理；若超出，检查可见性时间、outbox租约、扫描周期与队列等待之和。queued恢复包含60秒未领取阈值，不能拿租约恢复阈值直接套用；两条路径都要加入扫描周期、10秒退避（仅过期运行任务）和实际队列等待。

文件：verification-record.json

{
  "case": "publish_after_send_before_mark",
  "job_id": "填写本次实际ID",
  "observed_states": [],
  "duplicate_deliveries": null,
  "recovery_seconds": null,
  "artifact_sha256": null,
  "result": "not_run",
  "products": "实际演练后填写；不要用预计行为替代观测"
}

你应该看到什么

证据文件填写真实 observed_states、消息次数与 recovery_seconds；未做项保持 result=not_run，超出预算的项目有明确排查记录。

刚才发生了什么

恢复预算由多个等待相加，过期运行任务和一直未领取任务经过不同路径。把预期与实测分开，才能判断实现是否真的满足用户等待时间。

检查学习结果

• 每个故障案例有实际job_id、时间、状态和恢复证据才标通过。
• 成功文件与payload一致，失败或取消不会被旧token发布成功。
• 结果缓存过期不导致业务记录404，容量与调度去重也经过验证。
• 已sent未领取、退避重复消息与跨分钟周期重投都有独立案例，不被普通worker存活检查替代。
• 任务执行attempts不会因单纯重新派发增加；等待有持久时间和deadline边界。

进一步验证：约束与故障

先通过上面的学习结果检查，再做这一组实验。

准备把导出服务交给他人维护，需要证明故障可恢复且结果可追踪。

实现决策

• 先设受理、开始、完成和恢复的可量化目标，再配置阈值。
• 日志以业务ID关联，不把敏感商品内容当诊断材料。
• 把未执行的演练标为not_run，只有实际记录支持通过结论。

隔离环境演练

逐项执行本课矩阵，恢复所有本地组件与临时配置后，再跑一次无故障导出作为回归。

恢复验收

• 每项恢复结论都有观测证据。
• 已受理任务能定位到待发布、运行或明确终态。
• 不会把语法校验或eager函数测试描述为生产可靠性证明。
• 恢复覆盖queued和running两类滞留状态，时间槽去重范围与实际实现一致。

遇到问题时查这里

只有worker心跳正常就判定服务正常

原因：worker可能活着但队列积压、outbox不发布或任务卡住。

处理：同时观察最老等待、过期租约、失败和用户完成时间。

测试使用task_always_eager却声称broker恢复已验证

原因：eager在当前进程执行，绕过真实交付链路。

处理：eager只做函数级测试，消息重投与worker故障使用隔离集成环境。

自己试一次

选择“发布成功但sent_at未写回”案例，写出日志应包含的三个ID，并说明如何判断重复消息没有改变最终结果。

提示：区分业务job_id、outbox消息ID和租约token。

参考答案：

同一job_id关联重复投递日志，消息ID可能相同或不同；新领取有新token，只有有效token完成CAS。最终唯一result_ref指向与payload一致的文件，旧token完成返回False。

核心概念回顾

回想刚才的操作、现象与选择，用下面的概念解释你已经完成的案例。

指标要对应用户等待过程

完成第 2 步“查询积压、租约和失败”后，再回顾这条规则：排队数量之外，还要测受理到开始、运行时长、总完成时间、最老待发布年龄、失败率和重试次数。只看worker存活无法知道任务是否积压。

追踪业务ID与交付ID

完成第 1 步“给worker添加结构化阶段日志”后，再回顾这条规则：日志同时记录job_id、outbox_id或Celery task_id、lease_token和attempt，才能区分同一个任务的重复消息与重新执行。日志不打印完整商品描述、数据库密码或访问密钥。

验收包括故障后状态一致性

完成第 4 步“按矩阵逐个注入故障”后，再回顾这条规则：任务成功需要有效结果引用和可解析文件；取消或失败不能被旧worker覆盖。单元测试能检查状态转换，真实broker、worker和数据库故障窗口仍需隔离集成演练。

技术文档

• Celery监控
• Celery测试边界
• MySQL 8.4 Performance Schema