带租约的 SQLite 任务队列

功能说明

小型服务不一定需要独立的消息中间件，SQLite 的写事务和 WAL 模式足以支撑单机持久化队列。核心问题是多个消费者同时抢任务时不能重复处理。本例给任务行增加 locked_until 字段：消费者在一个写事务中挑选最早到期且未被锁定或租约已过期的任务，把锁定截止时间向后推，并返回任务内容。任务完成就删除，失败则清空锁并设置下次运行时间，超过租约仍未完成的任务会被其他消费者重新领取。

attempts 字段在每次领取时自增，可用于限制最大重试次数或计算退避时间。索引建在 run_at 上，让待运行任务的排序和筛选保持高效。

关键代码

connect 把 isolation_level 设为 None，改为显式控制事务，并启用 WAL 与 busy_timeout，降低并发读写时的锁等待失败。claim 先执行 BEGIN IMMEDIATE 取得写锁，再用一条带 RETURNING 的 UPDATE 原子地完成挑选与标记。子查询只选择 run_at 已到且锁为空的记录，按运行时间升序取一条。COMMIT 之后返回字典，异常路径执行 ROLLBACK 避免事务悬挂。finish 删除行，retry_later 重置锁并安排下次运行。

使用注意

SQLite 适合单机低并发，多实例共享同一个数据库文件并不合适。租约时间必须大于任务最长执行时间，否则任务会被重复领取，因此长任务应定期续租。RETURNING 语法需要 SQLite 3.35 及以上。生产环境应为 attempts 设置上限并把超过上限的任务移入死信表。任务负载最好只存标识或小型 JSON，大对象放文件或对象存储。进程崩溃后未完成的任务会在租约到期后自动恢复。

验证方式

运行脚本，第一行输出 claimed: send-email attempts: 1，说明任务被领取并计数。finish 后再次领取得到 resize-image，证明已完成任务不再出现。最后把该任务延后六十秒再调用 claim，输出 None，说明未到运行时间的任务不会被领取。可把 delay 改为零并重复领取，观察 attempts 递增。

代码示例

import sqlite3
import time
import uuid
from contextlib import contextmanager
from collections.abc import Iterator

SCHEMA = """
CREATE TABLE IF NOT EXISTS jobs (
    id TEXT PRIMARY KEY,
    payload TEXT NOT NULL,
    run_at REAL NOT NULL,
    locked_until REAL,
    attempts INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS jobs_ready ON jobs (run_at);
"""


@contextmanager
def connect(path: str) -> Iterator[sqlite3.Connection]:
    conn = sqlite3.connect(path, isolation_level=None, timeout=5)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=5000")
    try:
        yield conn
    finally:
        conn.close()


def enqueue(conn: sqlite3.Connection, payload: str, delay: float = 0.0) -> str:
    job_id = uuid.uuid4().hex
    conn.execute(
        "INSERT INTO jobs (id, payload, run_at) VALUES (?, ?, ?)",
        (job_id, payload, time.time() + delay),
    )
    return job_id


def claim(conn: sqlite3.Connection, lease: float = 30.0) -> dict[str, object] | None:
    now = time.time()
    conn.execute("BEGIN IMMEDIATE")
    try:
        row = conn.execute(
            """
            UPDATE jobs SET locked_until = ?, attempts = attempts + 1
            WHERE id = (
                SELECT id FROM jobs
                WHERE run_at <= ? AND (locked_until IS NULL OR locked_until < ?)
                ORDER BY run_at LIMIT 1
            )
            RETURNING id, payload, attempts
            """,
            (now + lease, now, now),
        ).fetchone()
        conn.execute("COMMIT")
    except Exception:
        conn.execute("ROLLBACK")
        raise
    return dict(row) if row else None


def finish(conn: sqlite3.Connection, job_id: str) -> None:
    conn.execute("DELETE FROM jobs WHERE id = ?", (job_id,))


def retry_later(conn: sqlite3.Connection, job_id: str, delay: float = 0.0) -> None:
    conn.execute(
        "UPDATE jobs SET locked_until = NULL, run_at = ? WHERE id = ?",
        (time.time() + delay, job_id),
    )


if __name__ == "__main__":
    with connect(":memory:") as conn:
        conn.executescript(SCHEMA)
        enqueue(conn, "send-email")
        enqueue(conn, "resize-image")
        first = claim(conn)
        print("claimed:", first["payload"], "attempts:", first["attempts"])
        finish(conn, str(first["id"]))
        second = claim(conn)
        print("next:", second["payload"])
        retry_later(conn, str(second["id"]), delay=60)
        print("waiting jobs:", claim(conn))