Python 定时任务如何避免重复执行：文件锁、任务状态与异常恢复

定时任务最容易出问题的场景，往往不是完全跑不起来，而是同一批数据被多个进程重复处理：调度触发器延迟重试之后，新的任务实例刚好又启动，或是上一次任务异常闪退，状态记录还卡在「运行中」没来得及更新。比较稳妥的思路是把防重逻辑拆成三层：启动时先抢占互斥锁，正式执行业务前先落盘任务状态，执行结束后靠结果和时间戳完成最终校验。

不要用“查到进程存在就直接跳过”作为唯一的防重门槛。让锁负责同一时间的进程互斥，让状态记录负责异常场景下的任务恢复，让结果校验负责判定本次任务到底有没有真正执行完成。

先把“重复执行”拆成三个现场

假设任务每十分钟扫描一次 inbox/，把新文件写入 archive/。重复执行可能来自三处：调度器重叠启动，旧进程被强制终止后留下半成品，或者任务成功了但状态写入失败。三种现场的修复点不同，不能只增加一个更长的间隔。

示例全程只用Python标准库和本地JSON文件实现，完全适合单机脚本、定时清理任务和中小型数据同步场景。多机器部署的时候，锁文件必须放在所有实例都能访问到的可靠共享存储上，或者直接换成数据库租约实现；本地文件锁不会自动变成分布式锁。

用独占锁挡住同一时刻的第二个进程

在类 Unix 系统环境下，可以让任务启动后先打开一个固定的专属文件，尝试获取非阻塞的独占锁。如果拿不到锁，就说明已经有别的实例正在处理同一份任务，本次直接记录跳过日志，不要再启动后续的业务逻辑。

from pathlib import Path
import fcntl

LOCK_PATH = Path("var/report.lock")

def acquire_lock():
    LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
    handle = LOCK_PATH.open("a+")
    try:
        fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        handle.close()
        return None
    return handle

lock_handle = acquire_lock()
if lock_handle is None:
    print("another run is active; skip")
else:
    try:
        print("lock acquired")
        # 在这里调用一次业务函数
    finally:
        fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN)
        lock_handle.close()

句柄必须一直保留到业务函数结束。只在函数入口短暂加锁，随后关闭句柄，等于把保护范围截断了。Windows 环境不要直接照搬 fcntl，应使用对应平台的文件锁实现，并把这项差异写进部署检查单。

状态文件要能回答“上次发生了什么”

锁只能告诉我们当前有没有别的实例在跑。我们还需要一份独立的任务状态记录，至少要保存批次号、当前执行阶段、开始时间、结束时间和处理数据条数。写入状态文件的时候要先写临时文件，再用原子替换操作覆盖正式的状态文件，避免进程在写入中途闪退留下损坏的半完整JSON文件。

import json
import os
import time
from pathlib import Path

STATE_PATH = Path("var/report-state.json")

def save_state(data):
    STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
    temp_path = STATE_PATH.with_suffix(".tmp")
    temp_path.write_text(
        json.dumps(data, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    os.replace(temp_path, STATE_PATH)

run_id = str(int(time.time()))
save_state({"run_id": run_id, "phase": "running", "started_at": time.time()})

“running”并不等于失败。启动新一轮时，先读取 started_at，再结合进程监控、输出目录和业务批次判断它是否已经失联。不要只按固定分钟数判死，因为任务耗时可能随数据量变化。

异常时保留失败证据，恢复时避免重复搬运

业务处理函数要把输入批次和输出结果直接绑定起来。你可以给每个待处理文件计算一个稳定的任务键：源文件的相对路径加上内容摘要。处理前先查询已经标记完成的任务键，处理成功之后再写入完成记录；就算后续状态文件出现回滚，已经落盘的结果记录也能阻止重复写入。

def run_once(items, completed_keys):
    processed = 0
    for item in items:
        task_key = item.key
        if task_key in completed_keys:
            continue
        write_one_result(item)
        completed_keys.add(task_key)
        processed += 1
    return processed

try:
    count = run_once(load_items(), load_completed_keys())
    save_state({"run_id": run_id, "phase": "success", "count": count,
                "finished_at": time.time()})
except Exception as exc:
    save_state({"run_id": run_id, "phase": "failed",
                "error_type": type(exc).__name__,
                "error": str(exc), "failed_at": time.time()})
    raise

这里的关键不是捕获所有异常后继续，而是记录证据后让失败可见。若 write_one_result 不是幂等操作，应先把临时结果写到独立目录，全部完成后再做一次原子切换；否则重跑时仍可能产生重复数据。

启动检查和结束验收各做一次

启动检查负责判断是否存在仍在运行的实例、是否有失联的 running 记录、上次失败对应哪个批次。结束验收则不要只看 Python 进程返回码，还要核对状态文件的 phase、输出数量和本轮 run_id 是否一致。

state = read_state()
if state and state.get("phase") == "running":
    age = time.time() - state["started_at"]
    print(f"previous run is still reported as running: {age:.0f}s")
    # 结合监控和输出证据决定是否人工确认

result = read_state()
if result["run_id"] != run_id or result["phase"] != "success" or result["count"] 

上线前至少完成三轮验证：同时并发启动两份任务进程，确认只有一份能进入业务执行逻辑；任务跑到一半手动终止进程，确认下一轮调度能识别到失联的异常状态；把同一份输入重复投递两次，确认已有的完成键不会让输出结果重复生成。

常见误区与适用边界

把进程列表当成任务状态

有可能出现进程已经退出但状态记录没清理的情况，也可能外层包装脚本还在运行但真正的业务逻辑已经执行完了。进程存活信息只能作为辅助判断依据，绝对不能代替实际的业务结果记录。

只加长调度间隔

单纯拉长调度间隔只能降低任务撞车的概率，没法处理网络延迟、手动补跑任务和异常重启这类场景。互斥锁、业务幂等和可追溯的状态记录这些核心逻辑还是得保留。

把失联任务直接判定为可重跑

执行重跑之前要先检查临时输出文件和业务侧的批次状态。如果外部系统已经收到请求但本地还没来得及写成功记录，盲目重跑很可能直接造成重复提交。没法确认状态的时候，应该转人工核对，或者直接用业务侧预先定义好的去重键做校验。

上线前的最小检查清单

• 锁文件所在的文件夹权限配置正确，锁句柄的生命周期覆盖完整业务执行流程。
• 状态写入采用临时文件加原子替换的方式，字段包含 run_id、phase 和各个关键节点的时间戳。
• 每条待处理输入都对应一个稳定的任务键，成功记录和业务输出结果可以互相交叉核对。
• 任务失败后会留存异常类型、所属批次和执行阶段，后续的恢复动作不会覆盖原始的故障证据。
• 单机锁、共享存储租约和数据库租约的适用边界，提前在部署文档里写清楚。

相关问题

文件锁能不能保证业务一定不重复？

不能。它只能保护能访问到同一个锁文件的进程，而且保护范围只覆盖锁的持有生命周期，业务幂等键仍然是防重复的最后一道防线。

状态文件应该多久清理一次？

不要按固定时间直接全量删除状态记录。至少保留最近若干次的成功和失败记录，再按批次分批归档，至少要保证单次故障复盘能找到对应的完整证据。

什么时候应该换成任务队列？

当你的任务需要多机并发、可观测重试、优先级调度和长时间积压处理能力时，本地锁加JSON状态记录的方案就不够用了，可以评估带租约机制和结果确认能力的任务队列或者专业调度系统。