任务模型与执行流程

任务状态要能回答：哪一次业务动作需要执行，谁正在执行，失败后能否接续。只保存“成功或失败”，不足以恢复投递中断、Worker 崩溃和租约过期。

任务定义、计划、实例与尝试

对象	表达什么	示例
任务定义	任务类型、处理器与参数约束	销售日报处理器及统计参数
调度计划	引用任务定义，按时间规则产生实例	每天 02:00 生成销售报表
执行实例	某一轮需要完成的工作	处理 2026-10-29 的销售数据
执行尝试	某个 Worker 的一次领取和处理	第一次超时，第二次重新处理

任务定义回答“做什么”，调度计划回答“按什么时间规则产生这类任务”，两者不是同义名称。同一任务定义可以绑定多个计划，也可以由业务事件或一次性请求直接创建实例。周期任务由计划不断产生实例；重试只增加尝试，不应生成一个失去关联的新业务实例。

实例的业务键应稳定，例如 sales-report:2026-10-29；尝试 ID 或领取代数则随重试变化。把尝试 ID 作为通知接口的幂等键，会让每次重试都成为一次新的发送。

下面以日报说明定义、计划与执行记录的关系：每天产生一个独立实例，同一天失败后重试，只增加这个实例的尝试记录。

图中的两次尝试属于同一个 10 月 29 日实例，按先后发生，不表示同时执行。业务幂等键沿用实例身份，不能随尝试次数变化。

直接领取模式的状态

下面采用“Worker 直接从任务表领取”的模型。后文再讨论独立消息队列带来的投递状态。

dead 是本专题约定的终态，不同框架也可能叫 failed 或使用死信表。关键是明确它不再自动重试，并保留失败原因。skipped 表示条件已经不成立，例如提醒对应的预约被改期；它与处理异常不同。

租约过期后，只有核对恢复条件且仍有尝试额度的任务才进入重试；外部结果不确定时，应先查询或对账。恢复过程见下文的租约说明及幂等、重试与故障恢复。

运行中的取消要走协作协议：记录取消请求，由处理器在安全位置检查。直接把运行记录改成 cancelled，不能证明旧 Worker 已停止。

Python 示例：直接领取任务表

这个示例使用 Python 3.12、psycopg 3 与 PostgreSQL 18。先准备本地测试库 task_demo 与同名账号，按数据库任务队列的表结构创建 background_tasks，并为账号授予表的读写权限和自增序列的使用权限。连接示例固定使用 127.0.0.1:5432，密码在启动时输入。

Worker 每次只领取一条到期任务，处理结束后带着本次 lease_token 确认结果。连接开启 autocommit，使领取 SQL 独立提交；业务处理发生在领取事务结束之后，不会一直占着行锁。

为观察状态变化，处理器模拟四种结果：正常完成、条件失效、暂时失败和参数错误。模拟分支不能代替真实业务判断；实际处理器应读取订单、会员等当前记录，再决定动作是否仍有效。

将下面保存为 worker.py，同目录的 demo_support.py 实现在折叠块中：

python

import time
from contextlib import closing

from demo_support import claim, finish, handle, open_db, recover_expired


def main():
    with closing(open_db()) as db:
        while True:
            recover_expired(db)
            task = claim(db)  # CTE 领取语句提交后，才进入业务处理。
            if task is None:
                time.sleep(1)
                continue
            error = None
            try:
                state = handle(task)
            except TimeoutError:
                state = ("retry_wait" if task["attempts"] < task["max_attempts"]
                         else "dead")
                error = "temporary_failure"
            except Exception:
                state, error = "dead", "handler_failed"
            if finish(db, task, state, error):
                print(task["business_key"], task["attempts"], state, flush=True)
            else:
                print("失去租约，未确认：", task["business_key"], flush=True)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        pass

demo_support.py：模拟处理器、连接、领取、确认与租约回收

python

from getpass import getpass

import psycopg
from psycopg.rows import dict_row


def handle(task):
    # 仅模拟结果，不调用通知、支付等外部接口。
    key = task["business_key"]
    if key == "obsolete":
        return "skipped"
    if key == "temporary" and task["attempts"] == 1:
        raise TimeoutError("模拟首次调用超时")
    if key == "invalid":
        raise ValueError("模拟不可重试的参数错误")
    return "succeeded"


def open_db():
    db = psycopg.connect(
        host="127.0.0.1", port=5432, dbname="task_demo", user="task_demo",
        password=getpass("本地测试库密码："),
        autocommit=True, row_factory=dict_row, connect_timeout=5,
    )
    for key in ("normal", "obsolete", "temporary", "invalid"):
        db.execute("""
            INSERT INTO background_tasks
                (task_type, business_key, scheduled_at, available_at, max_attempts)
            VALUES ('model_demo', %s, statement_timestamp(), statement_timestamp(), 3)
            ON CONFLICT (task_type, business_key) DO NOTHING
        """, (key,))
    return db


def claim(db):
    # 同一短事务内锁定候选并更新，跳过其他 Worker 已锁住的记录。
    return db.execute("""
        WITH candidate AS (
            SELECT id FROM background_tasks
            WHERE task_type = 'model_demo'
              AND state IN ('pending', 'retry_wait')
              AND available_at <= statement_timestamp()
              AND attempts < max_attempts
            ORDER BY available_at, id LIMIT 1
            FOR UPDATE SKIP LOCKED
        )
        UPDATE background_tasks AS t
        SET state = 'running', attempts = t.attempts + 1,
            lease_token = t.lease_token + 1,
            lease_until = statement_timestamp() + interval '60 seconds'
        FROM candidate AS c
        WHERE t.id = c.id
        RETURNING t.*
    """).fetchone()


def finish(db, task, state, error=None):
    cursor = db.execute("""
        UPDATE background_tasks
        SET state = %(state)s, lease_until = NULL, last_error = %(error)s,
            available_at = statement_timestamp() + (%(delay)s * interval '1 second'),
            finished_at = CASE WHEN %(state)s = 'retry_wait'
                               THEN NULL ELSE statement_timestamp() END
        WHERE id = %(id)s AND state = 'running'
          AND lease_token = %(token)s AND lease_until > statement_timestamp()
    """, {
        "state": state, "error": error,
        "delay": 2 if state == "retry_wait" else 0,
        "id": task["id"], "token": task["lease_token"],
    })
    return cursor.rowcount == 1


def recover_expired(db):
    # 这里只回收可安全重试的模拟动作，首次领取也占一次额度。
    db.execute("""
        WITH expired AS (
            SELECT id FROM background_tasks
            WHERE task_type = 'model_demo' AND state = 'running'
              AND lease_until <= statement_timestamp()
            ORDER BY lease_until, id LIMIT 100
            FOR UPDATE SKIP LOCKED
        )
        UPDATE background_tasks AS t
        SET state = CASE WHEN t.attempts < t.max_attempts
                         THEN 'retry_wait' ELSE 'dead' END,
            available_at = statement_timestamp() + interval '2 seconds',
            lease_until = NULL, last_error = 'lease_expired',
            finished_at = CASE WHEN t.attempts < t.max_attempts
                               THEN NULL ELSE statement_timestamp() END
        FROM expired AS e
        WHERE t.id = e.id
    """)

两个文件放在一个空目录，运行 uv run --no-project --python 3.12 --with 'psycopg[binary]>=3.2,<4' worker.py，按提示输入测试账号密码，按 Ctrl+C 退出。首次运行会看到下面的变化；重启沿用数据库记录，不会重新创建已完成的示例任务。示例仅领取和回收 task_type = 'model_demo' 的记录。

示例业务键	状态变化
`normal`	`pending → running → succeeded`
`obsolete`	`pending → running → skipped`
`temporary`	首次失败进入 `retry_wait`，2 秒后再次领取并成功
`invalid`	`pending → running → dead`，不再自动领取

这里没有独立调度器：到期查询承担时间筛选，Worker 负责处理，数据库负责保存状态。停止进程时尚未确认的任务，在租约到期后会被重新安排或转入 dead。

这个示例用于理解流程：没有续租、完整尝试历史或真实业务副作用，回收仅适用于这里可安全重试的模拟处理。真实外部调用结果不确定时，先查询或对账；业务写入与任务确认的事务边界也要另行设计。固定 2 秒重试只是便于观察，连接故障与领取提交结果不确定时仍需恢复处理。

SKIP LOCKED 适合多消费者竞争队列表，不保证严格 FIFO；原理与限制见 PostgreSQL 行锁文档。Python 的自动提交行为见 psycopg 事务管理，批量领取与数据库维护见数据库任务队列。

触发、就绪与业务判断

触发信号产生任务实例，调度器判断是否可以安排执行，Worker 在产生业务效果前核对当前事实。这三个判断需要分开：

判断	关心什么	未满足时怎样处理
是否产生任务	周期到达、事件发生或人工补跑	不创建本次实例
是否具备执行资格	可执行时间、依赖、资源额度	保留等待状态，之后重新判断
业务动作是否仍有效	当前状态、版本、授权与业务期限	跳过旧动作，或按业务规则继续等待

就绪可以是查询条件，不一定单独存为状态。具体实现也可以把部分依赖或资源判断交给 Worker，但仍需明确判断依据与恢复路径。

执行资格不保证业务动作一定成立。订单检查到期时可能已经付款，提醒开始发送时用户可能已经续费。领取租约只确定本次处理资格；旧 Worker 是否还能提交结果、外部动作是否会重复，仍需要 token、条件更新与业务幂等约束。

调度与执行的分工

一个带独立队列的系统可以划分为下面几部分。队列是可选组件，小规模系统也可以由 Worker 直接领取数据库中的到期任务。

组件	负责什么	需要留下的证据
业务服务	创建任务或发布事件	业务键、事件 ID、数据版本
任务存储	保存计划、执行状态和下一次可执行时间	持久化记录、状态变更
调度器	发现到期任务并安排投递	原计划时间、投递结果
队列	缓冲待处理消息并交给消费者	待消费数量、消费确认
Worker	领取任务、判断条件、执行动作	执行尝试、耗时、结果
恢复与运维逻辑	回收超时任务、重试、处理死信	失败原因、重放记录、告警

消息代理确认接收消息，与消费者确认处理完成，是两个不同的确认阶段；RabbitMQ 的发布确认与消费确认明确区分了这两层。因此，“已入队”和“已完成”需要分别记录。

下面用直接领取数据库任务的模式展示协作。示例假定业务变更与任务实例能在同一数据库事务中保存；跨系统提交需要另行处理可靠投递。

图中的唤醒只提示 Worker 尝试领取，不能授予执行权。即使提示丢失，实例仍在存储中，可以再次被发现；Worker 也可以自行轮询领取，将调度和执行放在同一个进程。业务结果与任务记录能否同事务提交、跨系统副作用如何核对，见数据库任务队列和幂等、重试与故障恢复。

增加队列后的投递状态

如果调度器先投递到独立队列，通常还需要区分 pending、queued 和 running，或者使用单独的投递记录：

pending：实例已持久化，尚未完成投递。
queued：投递获得确认，等待消费者领取。
running：消费者已取得处理资格并开始本次尝试。

队列可能在投递器更新记录前就分发消息，因此不能靠一次“先更新、再发送”串联保证正确性。需要 Outbox、可恢复的投递状态，以及消费者的条件领取，见事件驱动与异步任务。

实例记录与尝试记录

实例记录保存当前汇总状态；独立的尝试记录保存每次执行的历史：

位置	建议保存的内容
实例	类型、业务键、参数版本、原计划时间、下次可执行时间、当前状态
领取信息	Worker 标识、领取代数、租约截止时间
重试控制	已领取次数、最大尝试次数、业务截止时间
尝试历史	开始和结束时间、结果、错误类别、关联日志

本专题的 attempts 包含首次领取，max_attempts = 5 表示最多领取 5 次，最多重试 4 次。领取后立即崩溃也消耗一次额度，避免反复崩溃的任务无限回收。

简化实现可以先把最后一次错误放在实例表；如果需要审计与人工重放，再保留独立尝试历史。最后一次错误不能解释整个恢复过程。

租约、心跳与迟到结果

租约允许系统在 Worker 失联后重新分配任务，但租约过期不等于旧进程已经停止。进程暂停后恢复，就可能与新的 Worker 同时运行。

每次领取递增 lease_token，续租和完成确认都附带当前 token，并检查租约尚未过期：

sql

-- 参数由数据库驱动绑定；token 是本次领取返回的代数。
UPDATE background_tasks
SET state = 'succeeded',
    finished_at = statement_timestamp(),
    lease_until = NULL
WHERE id = :task_id
  AND state = 'running'
  AND lease_token = :lease_token
  AND lease_until > statement_timestamp();

更新不到记录时，Worker 已失去确认资格，需要停止后续操作并记录异常。这可以阻止旧尝试覆盖新尝试的任务状态；它不能自动阻止旧进程调用外部接口，后者仍需幂等或下游检查。

SQS 的可见性超时也体现了“领取后暂时隐藏、未确认时可再次投递”的机制。这里的 token 字段是本专题数据库模型的设计，不能直接等同于各消息系统的实现。

成功需要定义到业务层

保存通知请求、消息入队、供应商接受发送、用户收到通知，是不同结果。任务成功应对应处理器承诺的那一步，并在字段或事件中说明。

业务写入与任务确认能处于同一数据库事务时，可以一起提交；跨数据库或外部服务时，必须明确部分成功后如何查询和恢复。具体失败路径见幂等、重试与故障恢复。

选择实现方式的约束

需求	可以采用的起点	需要确认的代价
每天固定时间执行少量批处理	Cron 或 Kubernetes CronJob	执行重叠、补跑、结果追踪
业务记录到期后检查状态	持久化任务表和到期扫描	扫描压力、领取竞争、时效
业务变化后处理耗时操作	事件投递与 Worker	双写、重复、乱序、消费积压
需要独立扩缩容的异步处理	消息队列与 Worker	消息确认和业务提交的边界
大量跨服务的多步骤任务	进一步评估调度或工作流系统	状态迁移、恢复策略和运维成本

选型前先明确允许延迟多久、错过执行是否需要补跑、操作能否重试，以及最终结果由哪个系统确认。任务量、精度和恢复要求没有明确时，工具名称不足以决定方案。

参考资料

psycopg 3：安装与 binary 依赖
psycopg 3：事务管理与 autocommit
PostgreSQL 18：SELECT：锁与 SKIP LOCKED
Amazon SQS：Visibility timeout
RabbitMQ：Consumer Acknowledgements and Publisher Confirms

任务模型与执行流程 ​

任务定义、计划、实例与尝试 ​

直接领取模式的状态 ​

Python 示例：直接领取任务表 ​

触发、就绪与业务判断 ​

调度与执行的分工 ​

增加队列后的投递状态 ​

实例记录与尝试记录 ​

租约、心跳与迟到结果 ​

成功需要定义到业务层 ​

选择实现方式的约束 ​

参考资料 ​