做后端开发这些年我越来越喜欢把“重”的东西做轻。最近整理手头一个跑在本地的工具代号叫 rea全称是 Rapid Execution Assistant核心功能就是把一批耗时任务按规则排好队、依次执行、失败自动重试最后把结果回传给你。可能你第一反应是“这不就是任务队列框架的活吗”但当你只有一台小服务器、任务量也不大又不想为了一个异步队列引入一大堆中间件依赖时会明白 rea 这类轻量方案有多省心。这篇文章适合那些想把任务调度逻辑握在自己手里、不想被重型框架束缚的开发者也适合正在造内部工具轮子、却还没想清楚取舍的人。我会把设计思路、关键代码、典型踩坑全部摊开讲你既能看明白“为什么这么设计”也能直接照着落地。1. 整体设计为什么把一个调度器做得这么轻1.1 定位取舍什么时候不该上重框架先聊一个很现实的问题。很多团队一提到任务调度第一反应就是上 Celery 或者 XXL-JOB 这类成熟框架。但在实际场景里尤其是内部工具、个人服务器、边缘设备上任务量可能一天就几百条硬件配置也一般。这时候强行上一套重框架成本是肉眼可见的多维护一套 Broker、多写一堆配置、部署链路变长业务还没跑起来光环境问题就能折腾半天。rea 的定位就是“单机、无外部依赖、逻辑透明”。它不需要 Redis不需要独立 Worker 进程组把存储和调度都收敛在一个进程里。有人会问这不就是倒退吗我的看法是工具要匹配场景。你只需要一个能在本地稳定跑、能持久化、能重试的执行器那就没必要背着一个分布式中间件到处走。打个生活化比方你要给家里换一个灯泡不需要请一个装修队。rea 就是那个“家用工具箱”平时不占地方但拧螺丝、剪电线、测电压都能干。而 Celery 更像工程队适合盖楼但为换灯泡派工程队就离谱了。1.2 核心功能与模块划分rea 的功能边界非常克制核心就四个模块模块职责实现要点任务注册把普通函数注册为可调度任务用装饰器维护函数名到函数的映射表任务存储持久化任务元数据与状态SQLite 单文件存储不需要额外服务调度执行从存储中取任务、执行、更新状态线程池 条件变量控制任务取出节奏失败补偿对失败任务做指数退避重试统一在调度层处理不污染业务代码这四块各管各的互不纠缠。任务注册只管“这个函数能被执行”存储只管“任务落到哪”调度执行只管“该跑的跑起来”失败补偿只管“没跑成的怎么再来一次”。边界清楚后面加功能才不会改一处崩一处。1.3 设计时定的几条死规矩在设计 rea 的时候我给自己定了几条规矩回头看都是救命的东西第一单机优先。所有数据落本地 SQLite不允许为了某个功能引入网络依赖。这样即使断网任务也能排队等网络恢复后继续重试。第二持久化优先。任务一旦入队就必须能扛住进程崩溃。如果放在纯内存里进程一重启全没了那这个工具就没有存在意义。第三可观测性优先。每个任务必须有独立的 trace_id日志里必须能串起完整执行链。后面排查问题全靠这条链。第四透明可控。调度器本身不隐式改业务逻辑任务函数写成什么样就执行成什么样重试策略显式声明。这几条规矩定下来之后很多讨论就变得简单了。比如有人想加内存队列来提速我一句“进程重启任务丢不丢”就把他劝退了有人想在任务函数内部自己写重试我一句“那重试次数你在哪看”也让他打消了念头。2. 核心细节与实操要点2.1 任务模型字段远比你想的多任务表是 rea 的心脏表结构看起来简单但每个字段背后都对应一个实际问题。CREATE TABLE tasks ( task_id TEXT PRIMARY KEY, func_name TEXT NOT NULL, args TEXT NOT NULL DEFAULT [], kwargs TEXT NOT NULL DEFAULT {}, priority INTEGER NOT NULL DEFAULT 5, status TEXT NOT NULL DEFAULT pending, attempts INTEGER NOT NULL DEFAULT 0, max_attempts INTEGER NOT NULL DEFAULT 3, next_run_at REAL NOT NULL, trace_id TEXT NOT NULL, created_at REAL NOT NULL, updated_at REAL NOT NULL );task_id 是主键用 UUID 生成保证唯一。func_name 存的是“模块名:函数名”这样注册表查找不会歧义。args 和 kwargs 用 JSON 序列化存任务所需的参数都塞进去。priority 是优先级默认 5范围 1 到 10越小越紧急。status 是当前状态常见取值就四个pending排队中、running执行中、success成功、failed失败。如果重试还没到上限我会额外用 retry_wait 表示“在等待下次执行”这样调度器一进来就知道哪些任务此刻不该动。attempts 和 max_attempts 是重试计数器next_run_at 是“下次可执行时间”用 Unix 时间戳浮点数存储。created_at 和 updated_at 记录创建和更新时间既方便排查也方便后面做统计面板。我强调一下 trace_id它和执行任务本身可能完全无关但每个任务从入队到最终结束的整条生命周期日志都必须带上它。没有 trace_id线上出了问题你只能对着时间线猜那是灾难。2.2 优先级与公平性怎么平衡优先级调度有一个经典毛病高优先级任务不停插队低优先级任务会被饿死。rea 的解法不算复杂但很实用我称之为“分组 超时升级”。分组调度器优先扫描 priority 3 的任务取完再造 priority 7 的最后才是 8 到 10 的。超时升级如果某个任务的 next_run_at 已经超过当前时间加上一个“最长等待阈值”比如 600 秒无论优先级多低调度器都会把它临时提升到最高分组去跑。有个简单的计算公式阈值 base_wait_time (10 - priority) * extra_time。比如 base_wait_time 设 300 秒extra_time 设 60 秒priority 是 10 的任务阈值就是 300 秒priority 是 5 的任务阈值就是 300 60 * 5 600 秒。这个公式保证低优先级任务最多等 300 到 600 秒之间就会被强制捞起来执行绝对不会无限饿死。实际用下来这个策略对混合负载特别友好。紧急的通知类任务能立刻跑大批量的数据清洗任务会慢慢排队但绝不会因为优先级低就一直出不了头。2.3 重试与幂等最容易踩的坑重试是整个 rea 里最容易被误解的部分。新手往往会写“任务函数内部 try-except失败了再跑一次”这是不对的。规范做法是任务函数只负责自己的执行一旦抛出异常就把异常信息连同状态交给调度器调度器根据重试策略决定要不要把任务放回队列。重试间隔用指数退避delay min(base * 2 ^ attempts, max_delay)。base 通常设 10 秒attempts 是已重试次数。第一次失败等 10 秒第二次等 20 秒第三次等 40 秒封顶 120 秒。这么设计是给下游服务留出恢复时间避免风暴式重试把本来就脆弱的依赖打挂。幂等更关键。任务重试意味着同一件事可能被执行多次必须保证多次执行的结果一致。我的做法是强制每个任务函数接收 trace_id并由业务方把 trace_id 作为幂等键写入自己的存储。比如发短信任务先检查 trace_id 是否已经发过发过就不再发写文件任务先以 trace_id 建锁文件存在就直接跳过。还必须区分“可重试异常”和“不可重试异常”。参数错误、代码 Bug 这类重试一万次也没用不能重试下游超时、网络抖动、临时限流这类才值得重试。所以可重试异常用 RetryableError 包装其他异常直接标记失败。2.4 可观测性日志里必须带上 trace_idrea 的所有日志默认 JSON 结构化输出每条日志都带 task_id、trace_id、action、status、cost_ms 这些字段。启动时打印一条总览包含 worker 数量、队列长度、当前状态分布。为什么这么较真因为任务调度是异步的执行顺序和入队顺序可能完全不一样。如果日志是普通文本你根本没法把一个任务的完整生命周期连起来看。有了结构化日志你只需要grep trace_idxxx就能把入队、执行、重试、成功的全部路径拉出来。我还保留了一个轻量指标接口返回当前队列长度、running 数量、成功总数、失败总数、平均执行耗时。这个接口不依赖任何采集系统直接 curl 就能看运维排查起来非常快。3. 实操过程与核心环节实现3.1 环境准备与项目布局先用一个小项目把 rea 跑起来。依赖非常少就三个FastAPI提供 HTTP 入口、uvicorn启动服务、pydantic参数校验。SQLite 用 Python 内置的 sqlite3不需要额外安装。项目目录长这样rea/ __init__.py task.py queue.py worker.py storage.py main.py tests/ test_basic.pystorage.py 负责数据库初始化、连接、CRUDtask.py 负责函数注册queue.py 负责入队和取任务worker.py 负责执行循环main.py 暴露 HTTP 接口。每个文件职责单一改起来不心慌。虚拟环境准备好之后直接pip install fastapi uvicorn pydantic就够。真不需要装别的这是这个项目最舒服的地方。3.2 核心代码从任务注册到执行循环先看 storage.py 的连接管理。这里有个关键点SQLite 默认连接跟线程绑定线程池里执行任务时必须允许跨线程使用连接。# storage.py import sqlite3, time from contextlib import contextmanager DB_PATH rea.db contextmanager def get_conn(): conn sqlite3.connect(DB_PATH, check_same_threadFalse) conn.row_factory sqlite3.Row try: yield conn conn.commit() finally: conn.close() def init_db(): with get_conn() as conn: conn.execute( CREATE TABLE IF NOT EXISTS tasks ( task_id TEXT PRIMARY KEY, func_name TEXT NOT NULL, args TEXT NOT NULL DEFAULT [], kwargs TEXT NOT NULL DEFAULT {}, priority INTEGER NOT NULL DEFAULT 5, status TEXT NOT NULL DEFAULT pending, attempts INTEGER NOT NULL DEFAULT 0, max_attempts INTEGER NOT NULL DEFAULT 3, next_run_at REAL NOT NULL, trace_id TEXT NOT NULL, created_at REAL NOT NULL, updated_at REAL NOT NULL ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_status_next ON tasks(status, next_run_at) )check_same_threadFalse 这个参数必须写否则线程池里的 Worker 會直接报 “SQLite objects created in a thread can only be used in that same thread”。索引也很关键调度器每次扫描都按 status 和 next_run_at 过滤没有这个索引任务多的时候扫描会变慢。然后是 task.py负责把函数注册成任务# task.py import functools _task_registry {} def task(max_attempts3, default_priority5): def wrapper(func): func_name f{func.__module__}:{func.__name__} _task_registry[func_name] func func.max_attempts max_attempts func.default_priority default_priority return func return wrapper def get_func(func_name): return _task_registry.get(func_name)这里用“模块名:函数名”作为唯一键避免不同模块里同名函数互相覆盖。装饰器还把 max_attempts 和 default_priority 挂在函数对象上入队时如果调用方没显式指定就从函数定义里取。再来看 queue.py里面有入队和取任务两个核心操作# queue.py import json, time, uuid def enqueue(func_name, *args, priorityNone, max_attemptsNone, **kwargs): if get_func(func_name) is None: raise ValueError(ftask {func_name} not registered) task_id uuid.uuid4().hex trace_id uuid.uuid4().hex now time.time() with get_conn() as conn: conn.execute( INSERT INTO tasks (task_id, func_name, args, kwargs, priority, status, attempts, max_attempts, next_run_at, trace_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?, pending, 0, ?, ?, ?, ?, ?) , ( task_id, func_name, json.dumps(list(args)), json.dumps(kwargs), priority if priority is not None else get_func(func_name).default_priority, max_attempts if max_attempts is not None else get_func(func_name).max_attempts, now, trace_id, now, now, ) ) return task_id, trace_id入队只是插入一条记录不触发执行。取任务的时候要同时把状态从 pending 改成 running这一步必须用“条件更新”来避免同一个任务被多个 Worker 同时取走。def pop_next_task(): now time.time() with get_conn() as conn: conn.execute(PRAGMA busy_timeout5000) row conn.execute( SELECT * FROM tasks WHERE status pending AND next_run_at ? ORDER BY priority ASC, created_at ASC LIMIT 1 , (now,), ).fetchone() if row is None: return None cur conn.execute( UPDATE tasks SET statusrunning, updated_at? WHERE task_id? AND statuspending , (now, row[task_id]), ) if cur.rowcount 0: return None return row看到 update 语句里的AND statuspending了吗这是防止重复执行的关键。两个线程同时读到同一行时只有第一个能把状态改成 running第二个的 rowcount 会是 0直接返回 None 重新取下一条。worker.py 是调度核心用线程池加轮询实现# worker.py import threading, time, traceback from concurrent.futures import ThreadPoolExecutor class Worker: def __init__(self, worker_count4, poll_interval0.5): self._stop threading.Event() self._poll_interval poll_interval self._pool ThreadPoolExecutor(max_workersworker_count) def start(self): self._thread threading.Thread(targetself._loop, daemonTrue) self._thread.start() def _loop(self): while not self._stop.is_set(): task pop_next_task() if task is None: self._stop.wait(self._poll_interval) continue self._pool.submit(self._run_and_update, task) def _run_and_update(self, task): task_id task[task_id] trace_id task[trace_id] func_name task[func_name] start time.time() try: func get_func(func_name) args json.loads(task[args]) kwargs json.loads(task[kwargs]) ret func(*args, trace_idtrace_id, **kwargs) self._mark_success(task_id, ret, time.time() - start) except RetryableError: self._mark_retry(task_id, trace_id) except Exception: self._mark_failed(task_id, trace_id, traceback.format_exc())执行完以后根据异常类型决定是成功、重试还是失败。_mark_retry 里会更新 attempts如果 attempts 已经大于等于 max_attempts 就直接标记 failed否则计算 next_run_at 然后恢复成 pending。3.3 运行验证与参数调整启动入口 main.py 就三件事初始化数据库、启动 Worker、挂 HTTP 接口# main.py from fastapi import FastAPI from pydantic import BaseModel app FastAPI() init_db() worker Worker(worker_count4) worker.start() class EnqueueRequest(BaseModel): func_name: str args: list [] kwargs: dict {} priority: int | None None app.post(/enqueue) def enqueue_endpoint(req: EnqueueRequest): task_id, trace_id enqueue( req.func_name, *req.args, priorityreq.priority, **req.kwargs, ) return {task_id: task_id, trace_id: trace_id}启动命令也很简单uvicorn main:app --host 0.0.0.0 --port 8000验证一个简单任务# demo_task.py from rea.task import task from rea.exception import RetryableError import random task(max_attempts5, default_priority3) def send_notice(user_id, trace_idNone): print(fsend to {user_id} started, trace{trace_id}) if random.random() 0.3: raise RetryableError(simulate timeout) print(fsend to {user_id} done)然后通过接口手动入队curl -X POST http://127.0.0.1:8000/enqueue \ -H Content-Type: application/json \ -d {func_name:demo_task:send_notice,args:[1001],priority:1}日志里能看到入队、执行、成功或者 RetryableError 重试的完整链路。调参的重点是 worker_count 和 poll_interval。worker_count 不宜太大SQLite 并发写有上限我的建议是 2 到 8 之间poll_interval 默认 0.5 秒任务实时性要求高就调成 0.1要求不高可以放宽到 1 秒减少无谓扫描。4. 常见问题与排查技巧实录4.1 任务一直 pending 不走这是最常遇到的问题。第一反应是看 worker 是不是没启动第二反应是看任务表里的 next_run_at 和当前时间的关系。排查命令sqlite3 rea.db select task_id, status, next_run_at, datetime(now) from tasks;如果 next_run_at 比当前时间大很多说明重试退避还没到期这是正常的。如果 next_run_at 已经过去很久还是 pending看 worker 线程的日志——有没有周期性扫描动作还是已经被某个任务卡住了。还有一种可能线程池已经满了后面的任务都在排队等着执行。4.2 任务重复执行这个坑我是在并发场景下踩到的。同一个任务被两个 Worker 同时执行原因是取任务和更新状态之间没有用条件更新锁住。排查思路很直接检查 pop_next_task 里的 UPDATE 语句有没有带上statuspending条件以及更新后是否检查了 rowcount。如果你已经带上了条件但仍出现重复那就是幂等没做好。任务函数内部必须靠 trace_id 去重这个责任在业务方。我会在代码评审时反复强调调度器只能保证“尽量不重复”但真正“绝不重复”必须靠业务侧配合。4.3 SQLite 报 database is locked这个报错几乎每个用 SQLite 的人都会遇到。根本原因是并发写的时候锁冲突太频繁。rea 的解法有三板斧开启 WAL 模式PRAGMA journal_modeWAL;读写不互斥。设置 busy_timeoutPRAGMA busy_timeout5000;写冲突时等待而不是立刻失败。降低 worker_count并发写线程数降下来锁竞争自然减少。开启 WAL 之后我实测任务吞吐量提升明显至少不再频繁出现 “database is locked” 的红色告警。4.4 自查清单我整理了一张表每次集成 rea 之前照着过一遍能少踩很多坑检查项目正确做法错误做法任务注册函数用 task 装饰器模块能正常 import直接传函数对象重启后找不到异常分类可重试错误抛 RetryableError所有异常统一重试浪费次数幂等设计业务函数按 trace_id 去重不设幂等重试产生脏数据重试参数按业务场景设 max_attempts默认 3 次不够或太多Worker 生命周期随进程优雅退出事件置位直接 kill 进程任务状态卡在 running这张表我打印出来贴在工位上不是给新人看的是提醒我自己。因为人在写业务代码的时候最容易忽略的就是这些“看似不重要但出事就要命”的细节。5. 可扩展的方向与我的实际体会5.1 轻量框架可以怎么扩展rea 虽然定位轻量但扩展空间其实不小我列几个我觉得有实际价值的多进程执行器如果任务是 CPU 密集型单进程线程池发挥不了多核性能。可以把 worker.py 的执行部分改成 multiprocessing 进程池任务仍从 SQLite 里取但执行放到子进程。改造点主要是任务函数要能被 pickled以及执行结果要能序列化回来。定时调度给入队接口加一个run_at字段入队时 next_run_at 直接设为未来的时间戳。这样一来rea 就同时具备了定时任务能力。想做每天凌晨三点跑数据清洗只需要在启动时把这些定时任务入队一次。执行统计面板利用 metrics 接口配合一个简单的 HTML 页面展示队列长度、成功率和平均耗时。虽然不花哨但团队里有人问“任务跑得怎么样”时你直接甩一个链接过去比口说一百句都管用。5.2 用了几轮以后的真实感受rea 这个工具我实际用在内部的数据清洗和通知推送任务上跑了几个月稳定的确超出预期。最直接的收益是排障时间大幅缩短。以前任务出问题要翻好几个服务的日志才能拼出完整链路现在每条日志带 trace_id一条命令就全出来了这是最值回票价的地方。要说遗憾也有。轻量框架最大的敌人不是功能不够而是不断加需求。每加一个“顺便”功能复杂度就上升一截慢慢地就向重框架滑过去了。所以我给自己立了一个规矩rea 只吸收和维护核心能力其他需求一律通过外部脚本或独立服务来实现。这个边界守住了它才能一直保持“轻”。如果你也打算造一个类似的内部工具我的建议是一开始就把持久化、trace_id、条件更新这三件事做对后面所有扩展都会顺畅很多。如果这三件事偷懒了后面每一个新功能都会让你想重构。
做后端开发这些年我越来越喜欢把“重”的东西做轻。最近整理手头一个跑在本地的工具代号叫 rea全称是 Rapid Execution Assistant核心功能就是把一批耗时任务按规则排好队、依次执行、失败自动重试最后把结果回传给你。可能你第一反应是“这不就是任务队列框架的活吗”但当你只有一台小服务器、任务量也不大又不想为了一个异步队列引入一大堆中间件依赖时会明白 rea 这类轻量方案有多省心。这篇文章适合那些想把任务调度逻辑握在自己手里、不想被重型框架束缚的开发者也适合正在造内部工具轮子、却还没想清楚取舍的人。我会把设计思路、关键代码、典型踩坑全部摊开讲你既能看明白“为什么这么设计”也能直接照着落地。1. 整体设计为什么把一个调度器做得这么轻1.1 定位取舍什么时候不该上重框架先聊一个很现实的问题。很多团队一提到任务调度第一反应就是上 Celery 或者 XXL-JOB 这类成熟框架。但在实际场景里尤其是内部工具、个人服务器、边缘设备上任务量可能一天就几百条硬件配置也一般。这时候强行上一套重框架成本是肉眼可见的多维护一套 Broker、多写一堆配置、部署链路变长业务还没跑起来光环境问题就能折腾半天。rea 的定位就是“单机、无外部依赖、逻辑透明”。它不需要 Redis不需要独立 Worker 进程组把存储和调度都收敛在一个进程里。有人会问这不就是倒退吗我的看法是工具要匹配场景。你只需要一个能在本地稳定跑、能持久化、能重试的执行器那就没必要背着一个分布式中间件到处走。打个生活化比方你要给家里换一个灯泡不需要请一个装修队。rea 就是那个“家用工具箱”平时不占地方但拧螺丝、剪电线、测电压都能干。而 Celery 更像工程队适合盖楼但为换灯泡派工程队就离谱了。1.2 核心功能与模块划分rea 的功能边界非常克制核心就四个模块模块职责实现要点任务注册把普通函数注册为可调度任务用装饰器维护函数名到函数的映射表任务存储持久化任务元数据与状态SQLite 单文件存储不需要额外服务调度执行从存储中取任务、执行、更新状态线程池 条件变量控制任务取出节奏失败补偿对失败任务做指数退避重试统一在调度层处理不污染业务代码这四块各管各的互不纠缠。任务注册只管“这个函数能被执行”存储只管“任务落到哪”调度执行只管“该跑的跑起来”失败补偿只管“没跑成的怎么再来一次”。边界清楚后面加功能才不会改一处崩一处。1.3 设计时定的几条死规矩在设计 rea 的时候我给自己定了几条规矩回头看都是救命的东西第一单机优先。所有数据落本地 SQLite不允许为了某个功能引入网络依赖。这样即使断网任务也能排队等网络恢复后继续重试。第二持久化优先。任务一旦入队就必须能扛住进程崩溃。如果放在纯内存里进程一重启全没了那这个工具就没有存在意义。第三可观测性优先。每个任务必须有独立的 trace_id日志里必须能串起完整执行链。后面排查问题全靠这条链。第四透明可控。调度器本身不隐式改业务逻辑任务函数写成什么样就执行成什么样重试策略显式声明。这几条规矩定下来之后很多讨论就变得简单了。比如有人想加内存队列来提速我一句“进程重启任务丢不丢”就把他劝退了有人想在任务函数内部自己写重试我一句“那重试次数你在哪看”也让他打消了念头。2. 核心细节与实操要点2.1 任务模型字段远比你想的多任务表是 rea 的心脏表结构看起来简单但每个字段背后都对应一个实际问题。CREATE TABLE tasks ( task_id TEXT PRIMARY KEY, func_name TEXT NOT NULL, args TEXT NOT NULL DEFAULT [], kwargs TEXT NOT NULL DEFAULT {}, priority INTEGER NOT NULL DEFAULT 5, status TEXT NOT NULL DEFAULT pending, attempts INTEGER NOT NULL DEFAULT 0, max_attempts INTEGER NOT NULL DEFAULT 3, next_run_at REAL NOT NULL, trace_id TEXT NOT NULL, created_at REAL NOT NULL, updated_at REAL NOT NULL );task_id 是主键用 UUID 生成保证唯一。func_name 存的是“模块名:函数名”这样注册表查找不会歧义。args 和 kwargs 用 JSON 序列化存任务所需的参数都塞进去。priority 是优先级默认 5范围 1 到 10越小越紧急。status 是当前状态常见取值就四个pending排队中、running执行中、success成功、failed失败。如果重试还没到上限我会额外用 retry_wait 表示“在等待下次执行”这样调度器一进来就知道哪些任务此刻不该动。attempts 和 max_attempts 是重试计数器next_run_at 是“下次可执行时间”用 Unix 时间戳浮点数存储。created_at 和 updated_at 记录创建和更新时间既方便排查也方便后面做统计面板。我强调一下 trace_id它和执行任务本身可能完全无关但每个任务从入队到最终结束的整条生命周期日志都必须带上它。没有 trace_id线上出了问题你只能对着时间线猜那是灾难。2.2 优先级与公平性怎么平衡优先级调度有一个经典毛病高优先级任务不停插队低优先级任务会被饿死。rea 的解法不算复杂但很实用我称之为“分组 超时升级”。分组调度器优先扫描 priority 3 的任务取完再造 priority 7 的最后才是 8 到 10 的。超时升级如果某个任务的 next_run_at 已经超过当前时间加上一个“最长等待阈值”比如 600 秒无论优先级多低调度器都会把它临时提升到最高分组去跑。有个简单的计算公式阈值 base_wait_time (10 - priority) * extra_time。比如 base_wait_time 设 300 秒extra_time 设 60 秒priority 是 10 的任务阈值就是 300 秒priority 是 5 的任务阈值就是 300 60 * 5 600 秒。这个公式保证低优先级任务最多等 300 到 600 秒之间就会被强制捞起来执行绝对不会无限饿死。实际用下来这个策略对混合负载特别友好。紧急的通知类任务能立刻跑大批量的数据清洗任务会慢慢排队但绝不会因为优先级低就一直出不了头。2.3 重试与幂等最容易踩的坑重试是整个 rea 里最容易被误解的部分。新手往往会写“任务函数内部 try-except失败了再跑一次”这是不对的。规范做法是任务函数只负责自己的执行一旦抛出异常就把异常信息连同状态交给调度器调度器根据重试策略决定要不要把任务放回队列。重试间隔用指数退避delay min(base * 2 ^ attempts, max_delay)。base 通常设 10 秒attempts 是已重试次数。第一次失败等 10 秒第二次等 20 秒第三次等 40 秒封顶 120 秒。这么设计是给下游服务留出恢复时间避免风暴式重试把本来就脆弱的依赖打挂。幂等更关键。任务重试意味着同一件事可能被执行多次必须保证多次执行的结果一致。我的做法是强制每个任务函数接收 trace_id并由业务方把 trace_id 作为幂等键写入自己的存储。比如发短信任务先检查 trace_id 是否已经发过发过就不再发写文件任务先以 trace_id 建锁文件存在就直接跳过。还必须区分“可重试异常”和“不可重试异常”。参数错误、代码 Bug 这类重试一万次也没用不能重试下游超时、网络抖动、临时限流这类才值得重试。所以可重试异常用 RetryableError 包装其他异常直接标记失败。2.4 可观测性日志里必须带上 trace_idrea 的所有日志默认 JSON 结构化输出每条日志都带 task_id、trace_id、action、status、cost_ms 这些字段。启动时打印一条总览包含 worker 数量、队列长度、当前状态分布。为什么这么较真因为任务调度是异步的执行顺序和入队顺序可能完全不一样。如果日志是普通文本你根本没法把一个任务的完整生命周期连起来看。有了结构化日志你只需要grep trace_idxxx就能把入队、执行、重试、成功的全部路径拉出来。我还保留了一个轻量指标接口返回当前队列长度、running 数量、成功总数、失败总数、平均执行耗时。这个接口不依赖任何采集系统直接 curl 就能看运维排查起来非常快。3. 实操过程与核心环节实现3.1 环境准备与项目布局先用一个小项目把 rea 跑起来。依赖非常少就三个FastAPI提供 HTTP 入口、uvicorn启动服务、pydantic参数校验。SQLite 用 Python 内置的 sqlite3不需要额外安装。项目目录长这样rea/ __init__.py task.py queue.py worker.py storage.py main.py tests/ test_basic.pystorage.py 负责数据库初始化、连接、CRUDtask.py 负责函数注册queue.py 负责入队和取任务worker.py 负责执行循环main.py 暴露 HTTP 接口。每个文件职责单一改起来不心慌。虚拟环境准备好之后直接pip install fastapi uvicorn pydantic就够。真不需要装别的这是这个项目最舒服的地方。3.2 核心代码从任务注册到执行循环先看 storage.py 的连接管理。这里有个关键点SQLite 默认连接跟线程绑定线程池里执行任务时必须允许跨线程使用连接。# storage.py import sqlite3, time from contextlib import contextmanager DB_PATH rea.db contextmanager def get_conn(): conn sqlite3.connect(DB_PATH, check_same_threadFalse) conn.row_factory sqlite3.Row try: yield conn conn.commit() finally: conn.close() def init_db(): with get_conn() as conn: conn.execute( CREATE TABLE IF NOT EXISTS tasks ( task_id TEXT PRIMARY KEY, func_name TEXT NOT NULL, args TEXT NOT NULL DEFAULT [], kwargs TEXT NOT NULL DEFAULT {}, priority INTEGER NOT NULL DEFAULT 5, status TEXT NOT NULL DEFAULT pending, attempts INTEGER NOT NULL DEFAULT 0, max_attempts INTEGER NOT NULL DEFAULT 3, next_run_at REAL NOT NULL, trace_id TEXT NOT NULL, created_at REAL NOT NULL, updated_at REAL NOT NULL ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_status_next ON tasks(status, next_run_at) )check_same_threadFalse 这个参数必须写否则线程池里的 Worker 會直接报 “SQLite objects created in a thread can only be used in that same thread”。索引也很关键调度器每次扫描都按 status 和 next_run_at 过滤没有这个索引任务多的时候扫描会变慢。然后是 task.py负责把函数注册成任务# task.py import functools _task_registry {} def task(max_attempts3, default_priority5): def wrapper(func): func_name f{func.__module__}:{func.__name__} _task_registry[func_name] func func.max_attempts max_attempts func.default_priority default_priority return func return wrapper def get_func(func_name): return _task_registry.get(func_name)这里用“模块名:函数名”作为唯一键避免不同模块里同名函数互相覆盖。装饰器还把 max_attempts 和 default_priority 挂在函数对象上入队时如果调用方没显式指定就从函数定义里取。再来看 queue.py里面有入队和取任务两个核心操作# queue.py import json, time, uuid def enqueue(func_name, *args, priorityNone, max_attemptsNone, **kwargs): if get_func(func_name) is None: raise ValueError(ftask {func_name} not registered) task_id uuid.uuid4().hex trace_id uuid.uuid4().hex now time.time() with get_conn() as conn: conn.execute( INSERT INTO tasks (task_id, func_name, args, kwargs, priority, status, attempts, max_attempts, next_run_at, trace_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?, pending, 0, ?, ?, ?, ?, ?) , ( task_id, func_name, json.dumps(list(args)), json.dumps(kwargs), priority if priority is not None else get_func(func_name).default_priority, max_attempts if max_attempts is not None else get_func(func_name).max_attempts, now, trace_id, now, now, ) ) return task_id, trace_id入队只是插入一条记录不触发执行。取任务的时候要同时把状态从 pending 改成 running这一步必须用“条件更新”来避免同一个任务被多个 Worker 同时取走。def pop_next_task(): now time.time() with get_conn() as conn: conn.execute(PRAGMA busy_timeout5000) row conn.execute( SELECT * FROM tasks WHERE status pending AND next_run_at ? ORDER BY priority ASC, created_at ASC LIMIT 1 , (now,), ).fetchone() if row is None: return None cur conn.execute( UPDATE tasks SET statusrunning, updated_at? WHERE task_id? AND statuspending , (now, row[task_id]), ) if cur.rowcount 0: return None return row看到 update 语句里的AND statuspending了吗这是防止重复执行的关键。两个线程同时读到同一行时只有第一个能把状态改成 running第二个的 rowcount 会是 0直接返回 None 重新取下一条。worker.py 是调度核心用线程池加轮询实现# worker.py import threading, time, traceback from concurrent.futures import ThreadPoolExecutor class Worker: def __init__(self, worker_count4, poll_interval0.5): self._stop threading.Event() self._poll_interval poll_interval self._pool ThreadPoolExecutor(max_workersworker_count) def start(self): self._thread threading.Thread(targetself._loop, daemonTrue) self._thread.start() def _loop(self): while not self._stop.is_set(): task pop_next_task() if task is None: self._stop.wait(self._poll_interval) continue self._pool.submit(self._run_and_update, task) def _run_and_update(self, task): task_id task[task_id] trace_id task[trace_id] func_name task[func_name] start time.time() try: func get_func(func_name) args json.loads(task[args]) kwargs json.loads(task[kwargs]) ret func(*args, trace_idtrace_id, **kwargs) self._mark_success(task_id, ret, time.time() - start) except RetryableError: self._mark_retry(task_id, trace_id) except Exception: self._mark_failed(task_id, trace_id, traceback.format_exc())执行完以后根据异常类型决定是成功、重试还是失败。_mark_retry 里会更新 attempts如果 attempts 已经大于等于 max_attempts 就直接标记 failed否则计算 next_run_at 然后恢复成 pending。3.3 运行验证与参数调整启动入口 main.py 就三件事初始化数据库、启动 Worker、挂 HTTP 接口# main.py from fastapi import FastAPI from pydantic import BaseModel app FastAPI() init_db() worker Worker(worker_count4) worker.start() class EnqueueRequest(BaseModel): func_name: str args: list [] kwargs: dict {} priority: int | None None app.post(/enqueue) def enqueue_endpoint(req: EnqueueRequest): task_id, trace_id enqueue( req.func_name, *req.args, priorityreq.priority, **req.kwargs, ) return {task_id: task_id, trace_id: trace_id}启动命令也很简单uvicorn main:app --host 0.0.0.0 --port 8000验证一个简单任务# demo_task.py from rea.task import task from rea.exception import RetryableError import random task(max_attempts5, default_priority3) def send_notice(user_id, trace_idNone): print(fsend to {user_id} started, trace{trace_id}) if random.random() 0.3: raise RetryableError(simulate timeout) print(fsend to {user_id} done)然后通过接口手动入队curl -X POST http://127.0.0.1:8000/enqueue \ -H Content-Type: application/json \ -d {func_name:demo_task:send_notice,args:[1001],priority:1}日志里能看到入队、执行、成功或者 RetryableError 重试的完整链路。调参的重点是 worker_count 和 poll_interval。worker_count 不宜太大SQLite 并发写有上限我的建议是 2 到 8 之间poll_interval 默认 0.5 秒任务实时性要求高就调成 0.1要求不高可以放宽到 1 秒减少无谓扫描。4. 常见问题与排查技巧实录4.1 任务一直 pending 不走这是最常遇到的问题。第一反应是看 worker 是不是没启动第二反应是看任务表里的 next_run_at 和当前时间的关系。排查命令sqlite3 rea.db select task_id, status, next_run_at, datetime(now) from tasks;如果 next_run_at 比当前时间大很多说明重试退避还没到期这是正常的。如果 next_run_at 已经过去很久还是 pending看 worker 线程的日志——有没有周期性扫描动作还是已经被某个任务卡住了。还有一种可能线程池已经满了后面的任务都在排队等着执行。4.2 任务重复执行这个坑我是在并发场景下踩到的。同一个任务被两个 Worker 同时执行原因是取任务和更新状态之间没有用条件更新锁住。排查思路很直接检查 pop_next_task 里的 UPDATE 语句有没有带上statuspending条件以及更新后是否检查了 rowcount。如果你已经带上了条件但仍出现重复那就是幂等没做好。任务函数内部必须靠 trace_id 去重这个责任在业务方。我会在代码评审时反复强调调度器只能保证“尽量不重复”但真正“绝不重复”必须靠业务侧配合。4.3 SQLite 报 database is locked这个报错几乎每个用 SQLite 的人都会遇到。根本原因是并发写的时候锁冲突太频繁。rea 的解法有三板斧开启 WAL 模式PRAGMA journal_modeWAL;读写不互斥。设置 busy_timeoutPRAGMA busy_timeout5000;写冲突时等待而不是立刻失败。降低 worker_count并发写线程数降下来锁竞争自然减少。开启 WAL 之后我实测任务吞吐量提升明显至少不再频繁出现 “database is locked” 的红色告警。4.4 自查清单我整理了一张表每次集成 rea 之前照着过一遍能少踩很多坑检查项目正确做法错误做法任务注册函数用 task 装饰器模块能正常 import直接传函数对象重启后找不到异常分类可重试错误抛 RetryableError所有异常统一重试浪费次数幂等设计业务函数按 trace_id 去重不设幂等重试产生脏数据重试参数按业务场景设 max_attempts默认 3 次不够或太多Worker 生命周期随进程优雅退出事件置位直接 kill 进程任务状态卡在 running这张表我打印出来贴在工位上不是给新人看的是提醒我自己。因为人在写业务代码的时候最容易忽略的就是这些“看似不重要但出事就要命”的细节。5. 可扩展的方向与我的实际体会5.1 轻量框架可以怎么扩展rea 虽然定位轻量但扩展空间其实不小我列几个我觉得有实际价值的多进程执行器如果任务是 CPU 密集型单进程线程池发挥不了多核性能。可以把 worker.py 的执行部分改成 multiprocessing 进程池任务仍从 SQLite 里取但执行放到子进程。改造点主要是任务函数要能被 pickled以及执行结果要能序列化回来。定时调度给入队接口加一个run_at字段入队时 next_run_at 直接设为未来的时间戳。这样一来rea 就同时具备了定时任务能力。想做每天凌晨三点跑数据清洗只需要在启动时把这些定时任务入队一次。执行统计面板利用 metrics 接口配合一个简单的 HTML 页面展示队列长度、成功率和平均耗时。虽然不花哨但团队里有人问“任务跑得怎么样”时你直接甩一个链接过去比口说一百句都管用。5.2 用了几轮以后的真实感受rea 这个工具我实际用在内部的数据清洗和通知推送任务上跑了几个月稳定的确超出预期。最直接的收益是排障时间大幅缩短。以前任务出问题要翻好几个服务的日志才能拼出完整链路现在每条日志带 trace_id一条命令就全出来了这是最值回票价的地方。要说遗憾也有。轻量框架最大的敌人不是功能不够而是不断加需求。每加一个“顺便”功能复杂度就上升一截慢慢地就向重框架滑过去了。所以我给自己立了一个规矩rea 只吸收和维护核心能力其他需求一律通过外部脚本或独立服务来实现。这个边界守住了它才能一直保持“轻”。如果你也打算造一个类似的内部工具我的建议是一开始就把持久化、trace_id、条件更新这三件事做对后面所有扩展都会顺畅很多。如果这三件事偷懒了后面每一个新功能都会让你想重构。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
