5-5 定时任务幂等性设计

什么是幂等性？

幂等性指：同一操作执行多次，结果和执行一次相同。对于定时任务来说，就是"重跑不重复，失败可恢复"。

为什么幂等性重要？

• 网络错误导致任务中途失败，需要重跑
• 手动触发测试，不想产生重复数据
• 时区或时钟问题导致任务多次执行
• 部署迁移后需要补跑历史任务

三种幂等设计模式

1. 去重ID

import hashlib, sqlite3

def process_item(item_id: str, data: dict):
    db = sqlite3.connect('/tmp/processed.db')
    db.execute("CREATE TABLE IF NOT EXISTS processed (id TEXT PRIMARY KEY, ts REAL)")
    
    # 检查是否已处理
    if db.execute("SELECT 1 FROM processed WHERE id=?", (item_id,)).fetchone():
        print(f"已处理过 {item_id}，跳过")
        return
    
    # 执行处理
    do_process(data)
    
    # 标记为已处理
    import time
    db.execute("INSERT INTO processed VALUES (?,?)", (item_id, time.time()))
    db.commit()

2. 状态检查

# 在执行前检查状态，避免重复操作
def publish_report(date: str):
    report_file = f"/reports/{date}.md"
    
    # 如果今天的报告已经存在，跳过
    if os.path.exists(report_file):
        print(f"{date}的报告已存在，跳过")
        return
    
    # 生成并保存报告
    content = generate_report(date)
    with open(report_file, 'w') as f:
        f.write(content)

3. 覆盖写入

# 最简单：直接覆盖，不管之前有没有
# 适合结果确定性强的任务（每次生成结果一样）
def save_daily_metrics(date: str, data: dict):
    # 直接覆盖，即使已经存在
    with open(f"/metrics/{date}.json", 'w') as f:
        json.dump(data, f)

日报采集的幂等改造示例

#!/usr/bin/env bash
TODAY=$(date '+%Y-%m-%d')
REPORT_FILE="/reports/ai_daily_${TODAY}.md"

# 幂等检查：今天的报告是否已生成
if [ -f "$REPORT_FILE" ]; then
    echo "今日报告已存在，直接推送缓存内容"
    cat "$REPORT_FILE"
    exit 0
fi

# 生成新报告
CONTENT=$(python3 daily_report.py)
echo "$CONTENT"

# 保存缓存（下次重跑时直接用）
echo "$CONTENT" > "$REPORT_FILE"

幂等性不是可选项，是必须项。任何定时任务都会遇到需要重跑的时候。没有幂等保护的任务在重跑时会产生重复数据或副作用，这在生产环境中是严重问题。