Python 定时任务上云选型：从单机脚本到队列 Worker 的架构决策

很多 Python 脚本一开始只是本地定时跑一下：每天拉接口、生成 CSV、清理临时文件、同步订单状态。等脚本变成生产任务后，问题就来了：应该继续放在一台云主机的 cron 里，还是改成容器任务、队列 Worker，或者函数运行模式？

这篇文章不绑定某个云厂商，也不讨论价格细节，而是按架构决策的顺序来拆：先看业务负载，再看约束条件，然后比较几种方案，最后给出一套适合大多数 Python 定时任务的落地结构。

业务负载：这个 Python 任务到底重不重

先不要急着选服务。一个 Python 定时任务适合什么架构，取决于它的负载特征，而不是脚本语言本身。我们先把任务按几个维度量化。

• 触发频率：每小时一次、每分钟一次，还是需要秒级触发。
• 单次耗时：几秒结束，还是会跑 20 分钟以上。
• 数据规模：处理几十条记录，还是需要批量处理几十万行。
• 外部依赖：是否依赖数据库、对象存储、第三方接口或消息队列。
• 失败影响：失败后能否明天补跑，还是会影响订单、账单或通知。

如果任务每天跑一次、耗时短、失败可手动补跑，单机 cron 就够用。反过来，如果任务有大量 I/O、需要并发消费、失败要重试，就应该尽早拆成“调度”和“处理”两个部分。

约束条件：时间、状态和失败重试

上云之后，最容易被低估的是三类约束：运行时间、任务状态和失败重试。

运行时间决定任务是否适合短生命周期平台。比如数据导出要跑 30 分钟，就不适合放到有严格运行时间上限的环境里。任务状态决定能不能横向扩容，如果所有进度都写在本地文件里，换实例后就很难接续。失败重试决定业务可靠性：简单地重新跑一遍，可能会重复发通知或重复写数据。

可以用下面的 Python 伪代码把任务切成可追踪的批次：

from dataclasses import dataclass
from datetime import datetime

@dataclass
class TaskBatch:
    batch_id: str
    source: str
    started_at: datetime | None = None
    finished_at: datetime | None = None
    status: str = "pending"

def mark_running(batch: TaskBatch) -> TaskBatch:
    batch.started_at = datetime.utcnow()
    batch.status = "running"
    return batch

def mark_done(batch: TaskBatch) -> TaskBatch:
    batch.finished_at = datetime.utcnow()
    batch.status = "done"
    return batch

这段代码不是完整系统，只是提醒一点：任务状态要有地方记录。只要状态能被数据库或存储服务持久化，任务就更容易重试、补跑和审计。

方案对比：cron、容器任务、队列 Worker 和函数运行

常见方案可以分成四类。

如果你无法确定该选哪个，可以先问一个问题：任务失败后是否需要自动重试和追踪？如果答案是“需要”，队列 Worker 通常比单机 cron 更稳。

推荐架构：调度器只发任务，Worker 负责处理

对于生产环境里的 Python 定时任务，推荐把调度和业务处理拆开：调度器只负责按时间创建任务消息，Worker 从队列里取消息并处理，处理结果写入数据库或对象存储。

这套结构的关键好处是边界清楚：

• 调度器轻量：只负责生成任务，不跑重逻辑。
• 队列缓冲：任务量突然增加时，Worker 可以慢慢消费。
• 状态可查：每个任务都有 pending、running、done、failed 等状态。
• 失败可重试：失败任务可以重新入队，但要配合幂等键。

Worker 入口可以保持很简单：

def handle_message(message: dict) -> None:
    task_id = message["task_id"]
    idempotent_key = message["idempotent_key"]

    if already_done(idempotent_key):
        return

    mark_task_running(task_id)
    try:
        rows = load_source_rows(message["source"])
        result = build_report(rows)
        save_result(task_id, result)
        mark_task_done(task_id)
    except Exception as error:
        mark_task_failed(task_id, str(error))
        raise

这里最重要的是 idempotent_key。它用来判断同一个业务批次是否已经处理过，避免队列重试时重复写结果。

风险点：重复运行、超时和结果丢失

这类架构的风险集中在三处。

• 重复运行：队列至少投递一次时，同一消息可能被处理多次，所以需要幂等键。
• 超时中断：任务可能跑到一半失败，因此每个阶段都要有状态记录。
• 结果丢失：生成文件或报表后，应先写入可靠存储，再更新任务状态。

还有一个常见误区：把 Worker 数量开得越多越好。实际上，如果下游数据库或第三方接口有并发限制，Worker 扩得太快反而会造成限流和失败重试风暴。

落地清单：上线前必须检查的项目

最后给一份上线清单，适合把本地 Python 脚本迁移成云上任务时逐项确认。

• 任务是否有唯一批次号和幂等键。
• 任务状态是否写入数据库或可靠存储。
• 失败是否有重试上限，是否有人工补跑入口。
• Worker 并发数是否小于下游系统承载能力。
• 日志是否包含 task_id、batch_id 和耗时。
• 结果文件是否先落可靠存储，再标记任务完成。

总结一下：Python 定时任务上云，不是简单地把脚本放到另一台机器上。低风险脚本可以继续用 cron；依赖复杂可以用容器任务；需要重试、追踪和扩容时，优先考虑“调度器 + 队列 + Worker + 状态库”的结构。先把负载和约束说清楚，架构选择就会稳定很多。