异步任务、取消与断点续跑

更新 2026-10-05

正在读取学习记录…

长任务是在一次页面请求结束后还要继续执行的工作，例如对三份资料生成报告。接口先接收任务并返回任务标识，后台工作者逐项处理；用户通过任务标识查询进度，必要时取消。异步接收让用户不用一直保持连接，后台任务仍然需要保存状态。

断点续跑依赖游标，也就是“已完成到哪里”。处理完第一份资料后保存游标1，下次从第二份开始。取消是一种明确状态，工作者到达可停止的位置时不再执行下一项；已经完成的结果可以保留，以便再次恢复。

中断一次三份资料的处理

下面用SQLite保存任务和结果。DOCUMENTS 提供三份固定资料，每次 tick 读取当前资料并提取首句作为摘要，然后推进游标。固定流程先完成第一项、取消任务、再次尝试推进，再关闭并重新打开数据库，恢复任务后完成剩下两项。

数据库由程序在系统配置的临时目录下建立独立文件夹，每次运行结束清理。需要把实验放在F盘时，在当前PowerShell设置 $env:TEMP = $env:TMP = "F:\学习\l4-long-running" 后运行；这只影响本窗口启动的程序。重新打开同一数据库验证连接重建后的状态；它与整台机器断电的故障范围不同。

取消保留游标，重开从下一项继续，每项前检查取消；结果与游标一起提交 — 图1：每项前检查取消；结果与游标一起提交。

保存完整程序并运行

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-long-running" -Force
cd "F:\学习\l4-long-running"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

新建 F:\学习\l4-long-running 文件夹，把下面完整代码保存为 long_running_demo.py，然后在PowerShell运行：

cd "F:\学习\l4-long-running"
ls
python -X utf8 long_running_demo.py

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-long-running" -Force
cd "F:\学习\l4-long-running"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

"""用 SQLite 保存批处理游标，在取消后重新读取进度继续。"""
import sqlite3
from pathlib import Path
from tempfile import TemporaryDirectory

DOCUMENTS = (
    "退款通常三至五个工作日到账。申请需填写订单编号。",
    "生鲜商品不支持无理由退货。质量问题请联系客服。",
    "配送通常需要两天。节假日可能延后。",
)


class Job:
    def __init__(self, path):
        self.db = sqlite3.connect(path)
        self.db.executescript("""
            CREATE TABLE IF NOT EXISTS job (
                id TEXT PRIMARY KEY, cursor INTEGER, status TEXT);
            CREATE TABLE IF NOT EXISTS result (
                item INTEGER PRIMARY KEY, summary TEXT);
            INSERT OR IGNORE INTO job VALUES ('report-42', 0, 'running');
        """)
        self.db.commit()

    def state(self):
        return self.db.execute("SELECT cursor, status FROM job").fetchone()

    def tick(self):
        cursor, status = self.state()
        if status != "running":
            return status
        # 每次处理一项，在同一事务中提交结果和游标。
        item = cursor + 1
        # 从本项原文提取首句作为摘要，结果随资料内容变化。
        summary = DOCUMENTS[cursor].split("。", 1)[0] + "。"
        with self.db:
            self.db.execute("INSERT INTO result VALUES (?, ?)", (item, summary))
            self.db.execute("UPDATE job SET cursor=?, status=?",
                            (item, "completed" if item == 3 else "running"))
        return self.state()[1]

    def cancel(self):
        with self.db:
            self.db.execute("UPDATE job SET status='cancelled' WHERE status='running'")

    def resume(self):
        with self.db:
            self.db.execute("UPDATE job SET status='running' WHERE status='cancelled'")

    def close(self):
        self.db.close()


def run_demo():
    # 独立临时数据库使每次运行都有相同初始状态。
    with TemporaryDirectory(prefix="long-job-") as folder:
        path = Path(folder) / "job.db"
        job = Job(path)
        job.tick()
        lines = [f"第一轮: 游标={job.state()[0]}, 状态={job.state()[1]}"]
        job.cancel()
        job.tick()
        lines.append(f"取消后继续调用: 游标={job.state()[0]}, 状态={job.state()[1]}")
        job.close()
        job = Job(path)
        lines.append(f"重新打开数据库: 游标={job.state()[0]}, 状态={job.state()[1]}")
        job.resume()
        job.tick()
        job.tick()
        count = job.db.execute("SELECT count(*) FROM result").fetchone()[0]
        lines.append(f"恢复完成: 游标={job.state()[0]}, 状态={job.state()[1]}, 结果数={count}")
        job.close()
        return lines


if __name__ == "__main__":
    print("\n".join(run_demo()))

完整固定输出如下：

第一轮: 游标=1, 状态=running
取消后继续调用: 游标=1, 状态=cancelled
重新打开数据库: 游标=1, 状态=cancelled
恢复完成: 游标=3, 状态=completed, 结果数=3

输出里的游标为什么停在1

取消后的第二行仍是游标1，因为 tick 看到 cancelled 就返回，没有处理第二份资料。第三行重新打开数据库仍看到相同游标和取消状态，说明这两项信息来自数据库。最后一行恢复为 completed，游标3、结果数3，前面完成的摘要没有重复插入。

tick 把结果写入和游标推进放在同一个 with self.db 事务里。第一条语句保存摘要，第二条语句更新进度，两个操作一起提交；任意一条失败则一起撤销。游标若先提交、摘要后失败，恢复时会跳过缺失结果；摘要先提交、游标未提交，则会重复处理。

取消在哪里生效

程序在每一项开始前检查状态，因此取消后不会开始下一项。这叫协作式取消：工作者主动检查是否该停止。若一次网络请求正在执行，取消标记本身不能终止请求，应把取消信号传给客户端，并为单次调用设置超时。

已经执行的外部动作需要按服务契约处理。生成本地摘要可保留；已经扣款的任务则可能需要补偿。向用户显示“取消已接收”与“任务已停止”时，应对应不同状态，以免工作者仍执行时误报完成取消。

页面请求与后台执行怎样衔接

网站接收任务时保存任务行、生成稳定标识，再将工作交给队列。工作者按任务标识读取状态和游标；查询接口只读任务进度，不重新发起处理。恢复接口将可恢复的取消状态改回运行，并重新调度同一任务。

本例通过固定函数调用模拟调度，没有HTTP接收接口、后台进程和同时运行的工作者。它展示持久进度与取消边界；接服务后还要使用租约或条件更新保证同一游标只有一个工作者推进。任务结果也应与租户、用户权限绑定。

重启和外部副作用怎样核对

数据库事务能一起提交本地结果与游标，却不能包住远端模型、付款和邮件服务。若模型调用完成、结果尚未保存就退出，恢复后可能再次调用。对可重复生成的摘要可接受一次额外成本；对扣款则必须带业务幂等键或查询外部回执。

配套测试检查取消后重开、完成后重复推进、摘要确实来自资料正文，以及结果主键冲突时游标仍为0。长期保留的任务需要记录错误原因、尝试次数、最近更新时间和过期规则。可以沿Temporal 源码进一步查看工作流恢复与活动超时的实现。

改一处输入，核对机制变化

将 DOCUMENTS 第一项首句中的“三至五”改为“二至四”，运行 python -X utf8 long_running_demo.py。游标与结果数保持1、3，但数据库保存的第一份摘要随原文改变。

要直接查看摘要，在 run_demo 的 job.tick() 后、lines = ... 前临时加入 print(job.db.execute("SELECT summary FROM result WHERE item=1").fetchone()[0])。保存后重跑，会先显示“退款通常二至四个工作日到账。”。恢复原文并删除这条临时打印，回到四行输出。

运行配套测试

将下面测试保存为同一文件夹的 test_long_running_demo.py。它使用固定场景检查状态与业务结果，可以直接运行：

python -X utf8 -m unittest -v test_long_running_demo.py

完整测试：test_long_running_demo.py

import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from long_running_demo import Job


class JobTests(unittest.TestCase):
    def setUp(self):
        self.folder = TemporaryDirectory()
        self.path = Path(self.folder.name) / "job.db"
        self.job = Job(self.path)

    def tearDown(self):
        self.job.close()
        self.folder.cleanup()

    def test_cancel_and_reopen(self):
        self.job.tick()
        self.job.cancel()
        self.assertEqual(self.job.tick(), "cancelled")
        self.job.close()
        self.job = Job(self.path)
        self.assertEqual(self.job.state(), (1, "cancelled"))
        self.job.resume()
        self.job.tick()
        self.job.tick()
        self.assertEqual(self.job.state(), (3, "completed"))

    def test_completed_does_not_restart(self):
        for _ in range(3):
            self.job.tick()
        self.job.cancel()
        self.job.resume()
        self.job.tick()
        self.assertEqual(self.job.state(), (3, "completed"))
        self.assertEqual(self.job.db.execute("SELECT count(*) FROM result").fetchone()[0], 3)

    def test_summary_comes_from_document_content(self):
        self.job.tick()
        summary = self.job.db.execute("SELECT summary FROM result WHERE item=1").fetchone()[0]
        self.assertEqual(summary, "退款通常三至五个工作日到账。")

    def test_transaction_rolls_back_cursor_and_result(self):
        # 制造结果主键冲突，验证失败不会把游标推进到缺失结果之后。
        self.job.db.execute("INSERT INTO result VALUES (1, '预先冲突')")
        self.job.db.commit()
        with self.assertRaises(Exception):
            self.job.tick()
        self.assertEqual(self.job.state(), (0, "running"))


if __name__ == "__main__":
    unittest.main()

课后练习展开习题收起习题

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.0.1

选择题 · 先判断边界

关于“异步任务、取消与断点续跑”，哪种做法更稳妥？

对应知识点：L4-long-running

选择一个答案A. 先明确输入、证据和停止条件B. 默认输出永远正确C. 用一次成功推断全部场景

练习 2 / 4 · v1.0.0

简答题 · 解释失败

写出一个失败输入并说明记录字段。

对应知识点：L4-long-running

你的回答

练习 3 / 4 · v1.0.0

代码题 · 边界测试

补充正常、边界、失败三组输入和预期结果。

对应知识点：L4-long-running

你的代码或伪代码

练习 4 / 4 · v1.0.0

场景题 · 结果未知

证据不足时如何设计状态和人工接管？

对应知识点：L4-long-running

你的回答