python

《Python编程实战》13.3 定时任务与系统集成

脚本要有人在对的时间触发：本节讲透 cron 表达式与 crontab 写法、systemd timer 单元文件（Linux 配置，本机 macOS 未实测）、launchd plist，并用 fcntl 文件锁真跑防重入、用 state 文件做幂等、日志落盘。

Leeting Yan 8 分钟阅读 3650 字

本节目标：把脚本挂到系统的调度器上按点触发，读懂 cron 表达式，会写 crontab 与 systemd timer 单元，并用文件锁防重入、用状态文件做幂等、把日志可靠落盘。
适用版本：Python 3.12+（实测 3.14.6）；仅用标准库（fcntl、logging、subprocess）

13.3 定时任务与系统集成

前两节我们把「文件」和「文档」都自动化了，但脚本还得有人在对的时间按下回车。真到生产里，这个「人」是系统的调度器：Linux 上的 cron 与 systemd timer，macOS 上的 cron 与 launchd。这一节的难点不在「怎么写一行 crontab」，而在被调度器反复触发时，脚本怎么保持正确——不重入、可幂等、日志可追。

13.3.1 cron 表达式：五个字段

cron 表达式是五个用空格分隔的字段，从「分」到「周」：

┌───────── 分钟 (0-59)
│ ┌─────── 小时 (0-23)
│ │ ┌───── 日   (1-31)
│ │ │ ┌─── 月   (1-12)
│ │ │ │ ┌─ 星期 (0-7，0 和 7 都表示周日)
│ │ │ │ │
* * * * *   要执行的命令

常见写法：

表达式含义备注
0 3 * * *每天 03:00最常用的「日报」时刻
*/15 * * * *每 15 分钟*/n 是步进
0 */2 * * *每 2 小时整点
30 6 * * 1-5工作日 06:301-5 是范围（周一至周五）
0 0 1 * *每月 1 号 0 点
0 9,18 * * *每天 9 点、18 点逗号是枚举

两个反直觉的坑：

  • 「日」和「星期」是「或」关系，不是「与」。0 0 13 * 5 表示「每月 13 号或每周五」，而不是「13 号且是周五」。想要「且」，得在命令里自己再判一次。
  • cron 的环境极简：PATH 只有 /usr/bin:/bin 之类，~/.zshrc 里的环境变量、pyenv、虚拟环境的 activate 通通不生效。务必在 crontab 里写绝对路径，或在命令里显式 source 环境。

13.3.2 把脚本写成「可调度」的样子

一个能被调度器安全反复调用的脚本，要满足：入口明确、退出码有意义、日志落盘、默认幂等。crontab 用 crontab -e 编辑、crontab -l 查看，内容示例如下：

# crontab 内容示例（cron 环境极简，路径全写绝对）
SHELL=/bin/bash
PATH=/usr/local/bin:/usr/bin:/bin
# 每天 03:00 跑日报；日志追加到文件，错误也进同一个文件
0 3 * * * /Users/me/jobs/venv/bin/python /Users/me/jobs/daily.py >> /Users/me/logs/daily.log 2>&1

>> ... 2>&1 是 cron 的经典写法：把标准输出和标准错误都重定向到日志文件。少了它，脚本报错时错误信息会以邮件形式堆积在系统里（或者干脆消失），你什么都看不到。

脚本自身也要把日志管起来——用标准库 logging 落盘，并配轮转，避免日志无限增长：

import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path

D = Path("/tmp/python_book/scratch/13")
log = logging.getLogger("job")
log.setLevel(logging.INFO)
fh = RotatingFileHandler(D / "job.log", maxBytes=10_000, backupCount=3, encoding="utf-8")
fh.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
log.addHandler(fh)
log.addHandler(logging.StreamHandler())      # 同时打到 stderr，方便 cron 邮件

真实输出（两轮处理后的日志尾部）：

2026-10-09 12:01:49,376 INFO 跳过（已处理）: B
2026-10-09 12:01:49,376 INFO 跳过（已处理）: C
2026-10-09 12:01:49,376 INFO 处理: D

RotatingFileHandler 在单文件超过 maxBytes 时轮转，最多保留 backupCount 个历史文件（job.log.1、job.log.2…）。单进程写日志用它没问题；多进程/多机共写同一文件则会互相截断——那种场景要用 SysLogHandler 或集中式日志。

13.3.3 幂等：重跑不重复干活

调度器可能因为补跑、手动触发、失败重试而重复执行同一批任务。幂等的定义是：跑一次和跑 N 次结果一样。最朴素可靠的实现是「状态文件记录已完成项」：

import json
from pathlib import Path

STATE = Path("/tmp/python_book/scratch/13/state.json")

def load_state() -> dict:
    return json.loads(STATE.read_text()) if STATE.exists() else {"done": []}

def process(items: list[str]) -> None:
    state = load_state()
    done = set(state["done"])
    for it in items:
        if it in done:
            log.info("跳过（已处理）: %s", it)      # 已处理，直接跳过
            continue
        log.info("处理: %s", it)                     # 真正干活
        done.add(it)
    state["done"] = sorted(done)
    STATE.write_text(json.dumps(state, ensure_ascii=False))

真实输出（第一次处理 A/B/C，第二次处理 B/C/D）：

INFO 处理: A
INFO 处理: B
INFO 处理: C
INFO 跳过（已处理）: B
INFO 跳过（已处理）: C
INFO 处理: D

state.json 内容：

{"done": ["A", "B", "C", "D"]}

B、C 第二次被识别为「已处理」而跳过，只有 D 是新活。三个工程要点：

  • 状态写入要「原子」：直接 write_text 覆盖，若写到一半进程被杀，会留下半个 JSON。稳妥做法是写临时文件再 os.replace（同盘 rename 原子），保证状态文件要么是旧的、要么是新的，不会损坏。
  • 状态文件会越来越大：done 列表无限增长，几年后可能几百 MB。定期清理已完成项（比如只留最近 90 天），或改用「按任务日期分目录的完成标记文件」。
  • 幂等的粒度要对：以「业务键」（订单号、文件哈希）而非「行号」为去重键。行号会随数据变化而漂移，业务键才稳定。

13.3.4 任务锁：防止上一次还没跑完

如果任务耗时可能超过调度间隔（比如每 5 分钟跑一次、但一次要跑 8 分钟），必须防止重入——否则第 2 次启动时第 1 次还没结束，两个进程同时写同一份数据。

标准做法是文件锁。fcntl.flock 提供排他锁，且进程退出（含崩溃）时内核自动释放，不会像「pid 文件」那样留下死锁：

import fcntl, os, time
from pathlib import Path

LOCK = Path("/tmp/python_book/scratch/13/job.lock")

def run_with_flock() -> int:
    with LOCK.open("w") as f:
        try:
            fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)   # 非阻塞拿锁
        except OSError:
            print("另一个实例正在运行，本次跳过")           # 拿不到就退出
            return 1
        print("获得锁，开始干活")
        f.write(str(os.getpid()))
        f.flush()
        time.sleep(0.2)
        print("干完，释放锁")
        return 0

真实输出（单进程）：

获得锁，开始干活
干完，释放锁

跨进程竞争（一个后台进程持锁 2 秒，另一个进程尝试拿锁）：

child holding
跳过：另一个实例正在运行（未获得锁）

LOCK_NB（non-blocking）是关键：拿不到锁立刻返回而不是傻等——对定时任务来说，「跳过本次」通常比「排队等待」更合适。

另一种常见原语是原子创建 pid 文件，用 os.open 的 O_CREAT | O_EXCL 标志：

def acquire_pid_lock(path: Path) -> bool:
    try:
        fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
    except FileExistsError:
        return False
    os.write(fd, str(os.getpid()).encode())
    os.close(fd)
    return True

真实输出：

第一次: True
第二次: False

O_EXCL 保证「检查存在 + 创建」是一个原子操作，没有竞态。但它的弱点是：进程崩溃后锁文件不会自动消失，下次启动会被误判为「正在运行」。所以 pid 锁通常要配合「读 pid 判进程是否还活着」的清理逻辑，或者干脆用 flock（更省心）。选择建议：

方案崩溃自动释放跨机器适用
fcntl.flock是否（单机）单机定时任务首选
O_CREAT|O_EXCL pid 文件否（需手动清理）否需要「人为覆盖」语义时
Redis 分布式锁—是多机/多实例

13.3.5 systemd timer：Linux 的现代调度（本机未实测）

cron 古老但通用；Linux 上更现代的选择是 systemd timer：它和 service 配套，支持依赖、日志进 journald、开机错过还能补跑（Persistent=true）。配置分两个文件。

服务单元（/etc/systemd/system/daily.service）：

[Unit]
Description=Daily report job
After=network-online.target

[Service]
Type=oneshot
WorkingDirectory=/opt/jobs
ExecStart=/opt/jobs/venv/bin/python /opt/jobs/daily.py
# 失败重试；日志进 journald（用 journalctl -u daily 查看）
Restart=on-failure
RestartSec=30

定时器单元（/etc/systemd/system/daily.timer）：

[Unit]
Description=Run daily.service every day at 03:00

[Timer]
OnCalendar=*-*-* 03:00:00
Persistent=true

[Install]
WantedBy=timers.target

启用与查看：

sudo systemctl daemon-reload
sudo systemctl enable --now daily.timer
systemctl list-timers --all          # 查看下次触发时间
journalctl -u daily.service          # 查看运行日志

OnCalendar 的语法比 cron 直观：*-*-* 03:00:00 就是「每天 03:00」，还支持 Mon..Fri、hourly、daily 等简写。Persistent=true 解决 cron 的一个老大难——机器关机期间错过的任务，开机后会补跑一次。

本机未实测：当前环境是 macOS 26.3，systemctl 不存在（which systemctl 返回空）。以上单元文件是 Linux 上的标准配置，未在本机运行过。macOS 对应的是 launchd，见下一小节。

13.3.6 macOS 用 launchd：plist 配置

macOS 用 launchd 管理定时任务，配置是 XML 格式的 .plist。下面这份用本机 plutil -lint 实测校验通过：

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
  <key>Label</key>
  <string>com.example.daily</string>
  <key>ProgramArguments</key>
  <array>
    <string>/usr/bin/python3</string>
    <string>/Users/me/jobs/daily.py</string>
  </array>
  <key>StartCalendarInterval</key>
  <dict>
    <key>Hour</key><integer>6</integer>
    <key>Minute</key><integer>30</integer>
  </dict>
  <key>StandardOutPath</key>
  <string>/Users/me/logs/daily.out.log</string>
  <key>StandardErrorPath</key>
  <string>/Users/me/logs/daily.err.log</string>
</dict>
</plist>

校验与加载：

plutil -lint com.example.daily.plist       # 校验语法
cp com.example.daily.plist ~/Library/LaunchAgents/
launchctl load ~/Library/LaunchAgents/com.example.daily.plist
launchctl list | grep com.example            # 查看是否已加载

真实校验输出：

com.example.daily.plist: OK

StartCalendarInterval 对应 cron 的「定时」语义；StandardOutPath / StandardErrorPath 是 launchd 帮你做的重定向，省去 cron 里手写 >> ... 2>&1。注意 macOS 仍保留 crontab（本机 /usr/bin/crontab 存在），但 Apple 已不推荐，新项目用 launchd。

13.3.7 集成检查清单

把脚本真正挂上生产前，逐条过：

检查项做法
绝对路径命令、解释器、脚本、日志全写绝对路径
环境隔离用 venv 里的 python，或在命令里显式设 PATH
防重入flock 拿不到锁就退出，不排队
幂等以业务键去重，状态文件原子写入
日志RotatingFileHandler 落盘 + cron 重定向兜底
退出码成功 0、失败非 0，让调度器/监控能判定
失败可见接告警（邮件、webhook），别让失败静默

最后一行尤其重要：定时任务最危险的形态是「静默失败」——它每天照跑，只是每天都在报错，没人看日志。至少要让失败退出码被监控捕获，或把 CRITICAL 级日志推到告警渠道。

延伸阅读

小结

  • cron 是五字段表达式（分 时 日 月 周）；「日」与「周」是或关系，且 cron 环境极简，路径必须写绝对。
  • 可调度脚本要日志落盘（RotatingFileHandler）+ cron 重定向 >> ... 2>&1 兜底，并给出有意义的退出码。
  • 幂等靠业务键 + 状态文件；状态写入用「临时文件 + os.replace」保证原子性。
  • 防重入首选 fcntl.flock（进程退出自动释放），LOCK_NB 拿不到就跳过；pid 文件方案崩溃后需手动清理。
  • systemd timer 是 Linux 现代方案，OnCalendar 直观、Persistent=true 能补跑关机期错过的任务——本机 macOS 未实测。
  • macOS 用 launchd 的 .plist，本机 plutil -lint 校验通过；crontab 仍可用但不推荐。

到这里，「数据与自动化」这一部分收尾：我们从 pandas 数据处理走到文件、文档、定时调度。下一章换一种交付形态——把脚本包装成别人也能用的命令行工具，从 Click / Typer 的接口设计开始。

阅读导航：上一节：Excel / Word / PDF 自动化 · 下一节：Click / Typer 构建 CLI 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「python」更多文章

  1. 《Python高级编程》目录
  2. 《Python高级编程》11.3 PEP 流程与版本迁移策略
  3. 《Python高级编程》11.2 嵌入式与自由线程运行时