企业官网建设流程全解析

做后端开发这些年，我越来越喜欢把“重”的东西做轻。最近整理手头一个跑在本地的工具，代号叫 rea，全称是 Rapid Execution Assistant，核心功能就是把一批耗时任务按规则排好队、依次执行、失败自动重试，最后把结果回传给你。可能你第一反应是“这不就是任务队列框架的活吗”，但当你只有一台小服务器、任务量也不大，又不想为了一个异步队列引入一大堆中间件依赖时，会明白 rea 这类轻量方案有多省心。

这篇文章适合那些想把任务调度逻辑握在自己手里、不想被重型框架束缚的开发者，也适合正在造内部工具轮子、却还没想清楚取舍的人。我会把设计思路、关键代码、典型踩坑全部摊开讲，你既能看明白“为什么这么设计”，也能直接照着落地。

1. 整体设计：为什么把一个调度器做得这么轻

1.1 定位取舍：什么时候不该上重框架

先聊一个很现实的问题。很多团队一提到任务调度，第一反应就是上 Celery 或者 XXL-JOB 这类成熟框架。但在实际场景里，尤其是内部工具、个人服务器、边缘设备上，任务量可能一天就几百条，硬件配置也一般。这时候强行上一套重框架，成本是肉眼可见的：多维护一套 Broker、多写一堆配置、部署链路变长，业务还没跑起来，光环境问题就能折腾半天。

rea 的定位就是“单机、无外部依赖、逻辑透明”。它不需要 Redis，不需要独立 Worker 进程组，把存储和调度都收敛在一个进程里。有人会问，这不就是倒退吗？我的看法是：工具要匹配场景。你只需要一个能在本地稳定跑、能持久化、能重试的执行器，那就没必要背着一个分布式中间件到处走。

打个生活化比方：你要给家里换一个灯泡，不需要请一个装修队。rea 就是那个“家用工具箱”，平时不占地方，但拧螺丝、剪电线、测电压都能干。而 Celery 更像工程队，适合盖楼，但为换灯泡派工程队就离谱了。

1.2 核心功能与模块划分

rea 的功能边界非常克制，核心就四个模块：

这四块各管各的，互不纠缠。任务注册只管“这个函数能被执行”，存储只管“任务落到哪”，调度执行只管“该跑的跑起来”，失败补偿只管“没跑成的怎么再来一次”。边界清楚，后面加功能才不会改一处崩一处。

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_id

rea 的所有日志默认 JSON 结构化输出，每条日志都带 task_id、trace_id、action、status、cost_ms 这些字段。启动时打印一条总览，包含 worker 数量、队列长度、当前状态分布。

为什么这么较真？因为任务调度是异步的，执行顺序和入队顺序可能完全不一样。如果日志是普通文本，你根本没法把一个任务的完整生命周期连起来看。有了结构化日志，你只需要grep trace_id=xxx，就能把入队、执行、重试、成功的全部路径拉出来。

我还保留了一个轻量指标接口，返回当前队列长度、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.py

storage.py 负责数据库初始化、连接、CRUD；task.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_thread=False) 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_thread=False 这个参数必须写，否则线程池里的 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_attempts=3, default_priority=5): 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, priority=None, max_attempts=None, **kwargs): if get_func(func_name) is None: raise ValueError(f"task {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_timeout=5000") 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 status='running', updated_at=? WHERE task_id=? AND status='pending' """, (now, row["task_id"]), ) if cur.rowcount == 0: return None return row

看到 update 语句里的AND status='pending'了吗？这是防止重复执行的关键。两个线程同时读到同一行时，只有第一个能把状态改成 running，第二个的 rowcount 会是 0，直接返回 None 重新取下一条。

worker.py 是调度核心，用线程池加轮询实现：

# worker.py import threading, time, traceback from concurrent.futures import ThreadPoolExecutor class Worker: def __init__(self, worker_count=4, poll_interval=0.5): self._stop = threading.Event() self._poll_interval = poll_interval self._pool = ThreadPoolExecutor(max_workers=worker_count) def start(self): self._thread = threading.Thread(target=self._loop, daemon=True) 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_id=trace_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_count=4) 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, priority=req.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_attempts=5, default_priority=3) def send_notice(user_id, trace_id=None): print(f"send to {user_id} started, trace={trace_id}") if random.random() < 0.3: raise RetryableError("simulate timeout") print(f"send 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 语句有没有带上status='pending'条件，以及更新后是否检查了 rowcount。

如果你已经带上了条件但仍出现重复，那就是幂等没做好。任务函数内部必须靠 trace_id 去重，这个责任在业务方。我会在代码评审时反复强调：调度器只能保证“尽量不重复”，但真正“绝不重复”必须靠业务侧配合。

4.3 SQLite 报 database is locked

这个报错几乎每个用 SQLite 的人都会遇到。根本原因是并发写的时候锁冲突太频繁。rea 的解法有三板斧：

• 开启 WAL 模式：PRAGMA journal_mode=WAL;，读写不互斥。
• 设置 busy_timeout：PRAGMA busy_timeout=5000;，写冲突时等待而不是立刻失败。
• 降低 worker_count：并发写线程数降下来，锁竞争自然减少。

开启 WAL 之后，我实测任务吞吐量提升明显，至少不再频繁出现 “database is locked” 的红色告警。

4.4 自查清单

我整理了一张表，每次集成 rea 之前照着过一遍，能少踩很多坑：

这张表我打印出来贴在工位上，不是给新人看的，是提醒我自己。因为人在写业务代码的时候，最容易忽略的就是这些“看似不重要但出事就要命”的细节。

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、条件更新这三件事做对，后面所有扩展都会顺畅很多。如果这三件事偷懒了，后面每一个新功能都会让你想重构。