数据库任务队列

数据库任务队列让 Worker 直接领取到期记录，省去独立消息代理。业务数据与任务位于同一个数据库时，还可以在同一事务中创建它们。代价是轮询、领取和历史记录都会消耗数据库资源。

下面使用 PostgreSQL 18 展示一个设计示例：短事务领取，事务外处理，带租约确认，再由回收器处理失联任务。SQL 中的 :参数名 表示由驱动绑定的值，不是可以原样复制的通用 SQL 语法。

表结构与业务键

sql

CREATE TABLE background_tasks (
    id            bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    task_type     text NOT NULL,
    business_key  text NOT NULL,
    payload       jsonb NOT NULL DEFAULT '{}'::jsonb,
    scheduled_at  timestamptz NOT NULL,
    available_at  timestamptz NOT NULL,
    state         text NOT NULL DEFAULT 'pending'
                  CHECK (state IN (
                      'pending', 'running', 'retry_wait',
                      'succeeded', 'skipped', 'cancelled', 'dead'
                  )),
    attempts      integer NOT NULL DEFAULT 0 CHECK (attempts >= 0),
    max_attempts  integer NOT NULL DEFAULT 5 CHECK (max_attempts > 0),
    lease_token   bigint NOT NULL DEFAULT 0,
    lease_until   timestamptz,
    last_error    text,
    created_at    timestamptz NOT NULL DEFAULT statement_timestamp(),
    finished_at   timestamptz,
    UNIQUE (task_type, business_key)
);

CREATE INDEX background_tasks_due_idx
    ON background_tasks (available_at, id)
    WHERE state IN ('pending', 'retry_wait');

CREATE INDEX background_tasks_lease_idx
    ON background_tasks (lease_until, id)
    WHERE state = 'running';

周期任务的业务键可以包含计划时间，一次性任务可以包含订单 ID 与到期版本。首次创建时通常令 available_at = scheduled_at；重试只推进 available_at。

该最小模型没有单独保存尝试历史、取消请求和业务截止时间。需要这些能力时，应按任务模型扩展字段或关联表，不能仅凭最后一条错误恢复完整历史。

原子领取

Worker 只领取当前空闲执行槽能容纳的任务。下面的 batch_size 必须是正数，并受配置上限约束；60 秒租约只是示例。

sql

BEGIN;

WITH candidates AS (
    SELECT id
    FROM background_tasks
    WHERE state IN ('pending', 'retry_wait')
      AND available_at <= statement_timestamp()
      AND attempts < max_attempts
    ORDER BY available_at, id
    LIMIT :batch_size
    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 candidates AS c
WHERE t.id = c.id
RETURNING t.*;

COMMIT;

提交后再开始耗时处理，避免持有数据库行锁等待外部接口。SKIP LOCKED 跳过被其他事务锁住的候选记录，适合多个消费者领取队列表；它提供的视图不适合一般一致性查询，也不保证严格全局 FIFO。参见 PostgreSQL 行锁说明。

如果领取事务提交成功但 Worker 没收到响应，任务会暂时处于 running，之后由租约回收；不要因为调用超时就认为领取事务一定回滚。

续租与完成确认

长任务需要在租约到期前续租，续租失败就停止后续动作并记录失去所有权：

sql

UPDATE background_tasks
SET lease_until = statement_timestamp() + interval '60 seconds'
WHERE id = :task_id
  AND state = 'running'
  AND lease_token = :lease_token
  AND lease_until > statement_timestamp();

成功确认使用相同条件，把状态写成 succeeded 并设置 finished_at、清空租约，完整语句见任务模型与执行流程。

这里使用 statement_timestamp()，使一条语句使用同一个时间基准。PostgreSQL 的 CURRENT_TIMESTAMP / now() 取事务开始时间，长事务中可能不是当前墙上时间；区别见当前时间函数。

失败后安排重试

处理器先把错误归类，再计算带抖动的等待秒数。下面的 retryable 是布尔值，retry_delay_seconds 是有上限的非负数：

sql

UPDATE background_tasks
SET state = CASE
        WHEN :retryable AND attempts < max_attempts
            THEN 'retry_wait'
        ELSE 'dead'
    END,
    available_at = statement_timestamp()
        + (:retry_delay_seconds * interval '1 second'),
    finished_at = CASE
        WHEN :retryable AND attempts < max_attempts THEN NULL
        ELSE statement_timestamp()
    END,
    lease_until = NULL,
    last_error = :sanitized_error
WHERE id = :task_id
  AND state = 'running'
  AND lease_token = :lease_token
  AND lease_until > statement_timestamp();

错误文本要限长并脱敏。业务条件已经失效时应确认成 skipped，不进入重试循环；需要业务截止时间的任务，还要把时效判断加入重试决策。

回收过期租约

Worker 崩溃后无法主动安排重试，回收器负责把租约过期的记录转回可领取状态。下面示例每批最多处理 100 条：

sql

WITH expired AS (
    SELECT id
    FROM background_tasks
    WHERE 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 '5 seconds',
    lease_until = NULL,
    finished_at = CASE
        WHEN t.attempts < t.max_attempts THEN NULL
        ELSE statement_timestamp()
    END,
    last_error = 'lease_expired'
FROM expired AS e
WHERE t.id = e.id;

示例采用固定 5 秒恢复等待。大批任务同时回收时应分批并加入抖动，防止恢复流量集中冲击下游。即使任务没有机会主动报告失败，领取次数也会让它最终进入 dead。

Worker 的处理边界

text

按空闲槽位领取一批任务并提交
对每个任务并发执行：
  保存本次 lease_token，并启动受控的续租逻辑
  检查业务当前状态、版本与截止时间
  若已失效：带 token 确认 skipped
  否则执行业务动作
  成功：带 token 确认 succeeded
  失败：分类后带 token 安排 retry_wait 或 dead
  结束续租；失去租约时停止后续动作

以上是执行流程伪代码。真正的实现还要处理优雅退出、续租失败、连接池耗尽，以及已经开始但不能立即取消的外部调用。租约 token 保护任务记录；外部副作用仍按可靠性文章处理。

业务写入和完成确认如果在同一数据库事务中，应把所有权校验、业务条件检查与写入、任务确认一起设计；确认更新失败时回滚业务事务。跨系统调用无法靠这个事务覆盖。

容量与维护

轮询无结果时可以退避并加入抖动；积压期间按批处理，不要让所有 Worker 无间隔抢查。监控连接池等待、慢查询、锁等待、写入日志和表膨胀，历史任务按保留策略归档。

归档还要考虑业务键去重：删除任务记录会同时删除它承担的唯一约束。需要长期抵御重放的动作，应把业务幂等记录保留在独立位置。

如果任务扫描与状态更新已经明显影响在线事务，就需要评估隔离任务存储或引入独立队列；引入队列后，数据库与消息投递之间的恢复问题也要一起解决。

参考资料

PostgreSQL 18：SELECT：锁与 SKIP LOCKED
PostgreSQL 18：Current Date/Time

数据库任务队列 ​

表结构与业务键 ​

原子领取 ​

续租与完成确认 ​

失败后安排重试 ​

回收过期租约 ​

Worker 的处理边界 ​

容量与维护 ​

参考资料 ​