一、代码层面：Python 定时任务的 6 大方案

1. time.sleep() —— 最原始的方式

原理：在循环中使用 time.sleep() 暂停线程来实现“定时”效果。

适用场景：快速原型、脚本级简单延迟，严禁用于生产环境。

import time

def job():
    print("执行任务")

while True:
    job()
    time.sleep(60)  # 每隔60秒执行一次

缺点：

• 阻塞主线程，无法并发
• 程序崩溃即任务丢失
• 没有重试、超时、告警等机制
• 精度差（受 GIL 和调度影响）

---

2. schedule 库 —— 轻量级定时调度

原理：纯 Python 实现的任务调度器，语法简洁，类似 cron 表达式。

安装：pip install schedule

import schedule
import time

def job():
    print("每10秒执行一次")

schedule.every(10).seconds.do(job)
schedule.every().day.at("10:30").do(job)
schedule.every().monday.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

优点：语法直观，适合单进程小任务。
缺点：

• 仍是单线程阻塞
• 不支持持久化
• 不支持分布式
• 任务多了调度精度下降

---

3. APScheduler（Advanced Python Scheduler）—— 生产级调度库

核心概念：

组件	说明
Trigger	触发器，定义何时执行（date / interval / cron）
Job Store	任务存储，支持内存、SQLAlchemy、MongoDB、Redis
Executor	执行器，支持线程池、进程池
Scheduler	调度器，协调上述组件

安装：pip install apscheduler

3.1 基础示例（BlockingScheduler）

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger

def my_job(param):
    print(f"任务执行，参数：{param}")

scheduler = BlockingScheduler()

# interval 触发器：每30秒执行
scheduler.add_job(my_job, 'interval', seconds=30, args=['hello'])

# cron 触发器：每天凌晨2点执行
scheduler.add_job(my_job, CronTrigger(hour=2, minute=0), args=['daily'])

# date 触发器：指定时间执行一次
scheduler.add_job(my_job, 'date', run_date='2025-01-01 00:00:00', args=['once'])

scheduler.start()

3.2 持久化任务（SQLAlchemy JobStore）

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor

jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
    'default': ThreadPoolExecutor(max_workers=10)
}
job_defaults = {
    'coalesce': True,        # 错过多次只执行一次
    'max_instances': 3,      # 同一任务最大并发实例数
    'misfire_grace_time': 60  # 错过执行的宽限秒数
}

scheduler = BackgroundScheduler(
    jobstores=jobstores,
    executors=executors,
    job_defaults=job_defaults
)
scheduler.start()

关键参数说明：

• coalesce=True：如果任务因服务宕机错过了多次执行，恢复后只执行一次，而不是补执行所有错过的。
• max_instances=3：防止同一任务并发过多导致资源耗尽。
• misfire_grace_time=60：如果任务错过触发时间超过60秒，则放弃本次执行。

---

4. celery beat —— 分布式异步任务的定时调度

架构角色：

┌─────────────┐      ┌──────────────┐      ┌─────────────┐
│  Celery     │─────▶│  Message     │─────▶│  Celery     │
│  Beat       │      │  Broker      │      │  Worker     │
│  (定时器)   │      │ (Redis/RabbitMQ)│ │  (执行器)    │
└─────────────┘      └──────────────┘      └─────────────┘

安装：pip install celery redis

# celery_app.py
from celery import Celery
from celery.schedules import crontab

app = Celery(
    'tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',
    include=['tasks']
)

app.conf.beat_schedule = {
    'add-every-30-seconds': {
        'task': 'tasks.add',
        'schedule': 30.0,
        'args': (16, 7),
    },
    'daily-report': {
        'task': 'tasks.generate_report',
        'schedule': crontab(hour=2, minute=0),
    },
}
app.conf.timezone = 'Asia/Shanghai'

# tasks.py
from celery_app import app

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def add(self, x, y):
    try:
        return x + y
    except Exception as exc:
        raise self.retry(exc=exc)

@app.task
def generate_report():
    return "report generated"

启动命令：

celery -A celery_app worker --loglevel=info
celery -A celery_app beat --loglevel=info

---

5. threading.Timer —— 线程级定时

import threading

def delayed_task():
    print("5秒后执行")

t = threading.Timer(5.0, delayed_task)
t.start()

适用场景：单次延迟执行，不适合复杂调度。

---

6. asyncio 异步定时任务

import asyncio

async def periodic_task():
    while True:
        print("异步定时任务执行")
        await asyncio.sleep(10)

async def main():
    task = asyncio.create_task(periodic_task())
    await asyncio.sleep(60)  # 运行60秒后取消
    task.cancel()

asyncio.run(main())

---

二、中间件层面：消息队列 + 调度中间件

1. Redis —— 轻量级任务队列

使用 redis + rq-scheduler 或 apscheduler 结合 Redis 作为 Broker。

from rq import Queue
from rq_scheduler import Scheduler
from datetime import datetime, timedelta
from redis import Redis

scheduler = Scheduler(queue_name='default', connection=Redis())
scheduler.enqueue_at(datetime.now() + timedelta(minutes=5), my_func)

2. RabbitMQ —— 企业级消息队列

Celery 的默认推荐 Broker，支持消息确认、持久化、优先级队列，适合金融级场景。

3. Kafka —— 高吞吐事件流

适合大数据量的定时事件驱动场景，通常配合流式计算框架使用。

4. 方案对比总结

方案	持久化	分布式	重试	监控	适用规模
time.sleep	❌	❌	❌	❌	个人脚本
schedule	❌	❌	❌	❌	小型项目
APScheduler	✅	⚠️（需配合）	⚠️	⚠️	中型项目
Celery Beat	✅	✅	✅	✅	企业级
Airflow	✅	✅	✅	✅	数据流水线
XXL-Job	✅	✅	✅	✅	微服务集群

---

三、企业级任务调度架构设计

企业级任务调度平台架构图

┌──────────────────────────────────────────────────────────────────────────────────┐
│                               🏢 企业级任务调度平台                              │
├──────────────────────────────────────────────────────────────────────────────────┤
│                                                                                  │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐                       │
│  │  Web控制台    │    │  OpenAPI     │    │  CLI工具     │                       │
│  │  (任务管理UI) │    │  (RESTful)   │    │  (运维操作)   │                       │
│  └──────┬───────┘    └──────┬───────┘    └──────┬───────┘                       │
│         └────────────────────┼────────────────────┘                              │
│                              ▼                                                 │
│  ┌──────────────────────────────────────────────────────────────────────────┐  │
│  │                     🔧 调度管理层 (Scheduler Manager)                      │  │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────┐           │  │
│  │  │ 任务注册  │ │ 依赖编排  │ │ 优先级管理│ │ 分片策略  │ │ 限流   │           │  │
│  │  └──────────┘ └──────────┘ └──────────┘ └──────────┘ └────────┘           │  │
│  └──────────────────────────────┬───────────────────────────────────────────┘  │
│                                 ▼                                              │
│  ┌──────────────────────────────────────────────────────────────────────────┐  │
│  │                     ⏰ 调度引擎层 (Scheduling Engine)                      │  │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐             │  │
│  │  │  Master节点   │──│  ZooKeeper/   │──│  时间轮/CRON解析器    │             │  │
│  │  │  (调度决策)   │  │  ETCD选主     │  │  (触发计算)          │             │  │
│  │  └──────────────┘  └──────────────┘  └──────────────────────┘             │  │
│  │         │                                                                 │  │
│  │  ┌──────┴──────┐                                                         │  │
│  │  │ 任务分片路由 │ ── 一致性哈希 / 轮询 / 亲和性调度                        │  │
│  │  └─────────────┘                                                         │  │
│  └──────────────────────────────┬───────────────────────────────────────────┘  │
│                                 ▼                                              │
│  ┌──────────────────────────────────────────────────────────────────────────┐  │
│  │                     📨 消息中间件层 (Message Broker Layer)                 │  │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐             │  │
│  │  │  Redis       │  │  RabbitMQ    │  │  Kafka (可选)        │             │  │
│  │  │  (高频小任务) │  │  (可靠投递)   │  │  (事件流)           │             │  │
│  │  └──────────────┘  └──────────────┘  └──────────────────────┘             │  │
│  │         │                        │                                         │  │
│  │  ┌──────┴──────┐          ┌─────┴──────┐                                   │  │
│  │  │ 死信队列DLX  │          │ 延迟队列    │                                   │  │
│  │  │ (失败重试)   │          │ (定时触发)  │                                   │  │
│  │  └─────────────┘          └────────────┘                                   │  │
│  └──────────────────────────────┬───────────────────────────────────────────┘  │
│                                 ▼                                              │
│  ┌──────────────────────────────────────────────────────────────────────────┐  │
│  │                     ⚙️ 执行器层 (Worker Cluster)                           │  │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────────┐              │  │
│  │  │ Worker-1  │ │ Worker-2  │ │ Worker-3  │ │  ... Worker-N    │              │  │
│  │  │ (进程池)   │ │ (进程池)  │ │ (进程池)  │ │  (Auto Scaling) │              │  │
│  │  └──────────┘ └──────────┘ └──────────┘ └──────────────────┘              │  │
│  │         │                        │                                         │  │
│  │  ┌──────┴──────┐          ┌─────┴──────┐                                   │  │
│  │  │ 超时控制     │          │ 资源隔离    │                                   │  │
│  │  │ (软/硬超时)  │          │ (cgroup)   │                                   │  │
│  │  └─────────────┘          └────────────┘                                   │  │
│  └──────────────────────────────┬───────────────────────────────────────────┘  │
│                                 ▼                                              │
│  ┌──────────────────────────────────────────────────────────────────────────┐  │
│  │                     🛡️ 可观测性层 (Observability)                          │  │
│  │  ┌──────────────┐ ┌──────────────┐ ┌────────────────────────┐             │  │
│  │  │ Prometheus   │ │  Grafana     │ │  ELK 日志平台          │             │  │
│  │  │ (指标采集)    │ │ (可视化面板)  │ │  (日志聚合检索)        │             │  │
│  │  └──────────────┘ └──────────────┘ └────────────────────────┘             │  │
│  │  ┌──────────────┐ ┌──────────────┐ ┌────────────────────────┐             │  │
│  │  │ AlertManager │ │  链路追踪     │ │  审计日志              │             │  │
│  │  │ (告警路由)    │ │  (Jaeger)    │ │  (操作留痕)           │             │  │
│  │  └──────────────┘ └──────────────┘ └────────────────────────┘             │  │
│  └──────────────────────────────┬───────────────────────────────────────────┘  │
│                                 ▼                                              │
│  ┌──────────────────────────────────────────────────────────────────────────┐  │
│  │                     🔔 告警通知层 (Alerting)                              │  │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────────┐              │  │
│  │  │ 企业微信  │ │  钉钉     │ │  邮件     │ │  Webhook         │              │  │
│  │  │ 机器人    │ │  机器人   │ │  SMTP     │ │  自定义回调      │              │  │
│  │  └──────────┘ └──────────┘ └──────────┘ └──────────────────┘              │  │
│  └──────────────────────────────────────────────────────────────────────────┘  │
│                                                                                  │
│  ┌──────────────────────────────────────────────────────────────────────────┐  │
│  │                     💾 持久化层 (Persistence)                              │  │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐             │  │
│  │  │  MySQL/PG    │  │  Redis       │  │  Elasticsearch       │             │  │
│  │  │ (任务定义/    │  │ (分布式锁/    │  │ (执行日志/           │             │  │
│  │  │  执行记录)    │  │  状态缓存)    │  │  历史归档)           │             │  │
│  │  └──────────────┘  └──────────────┘  └──────────────────────┘             │  │
│  └──────────────────────────────────────────────────────────────────────────┘  │
└──────────────────────────────────────────────────────────────────────────────────┘

架构设计要点说明

层级	核心职责	关键技术选型
接入层	任务创建、管理、触发	FastAPI / Flask + JWT
调度管理层	任务编排、依赖解析、优先级	DAG 引擎、拓扑排序
调度引擎层	时间触发、选主、分片	ZooKeeper/ETCD、时间轮算法
消息中间件层	任务分发、削峰填谷	Redis Streams / RabbitMQ
执行器层	并发执行、超时控制	Celery Worker / 自研进程池
可观测性层	指标、日志、追踪	Prometheus + Grafana + ELK
告警通知层	多渠道告警推送	AlertManager + Webhook
持久化层	任务定义、执行历史	MySQL + Redis + ES

---

四、任务失败告警 + Prometheus 指标（完整代码）

1. 项目结构

scheduler_system/
├── app/
│   ├── __init__.py
│   ├── config.py
│   ├── metrics.py          # Prometheus 指标定义
│   ├── scheduler.py        # 调度引擎
│   ├── worker.py           # 任务执行器
│   ├── alerting.py         # 告警模块
│   └── models.py           # 数据模型
├── docker-compose.yml
├── prometheus.yml
├── alert_rules.yml
└── main.py

2. Prometheus 指标定义 (metrics.py)

"""
Prometheus 指标定义模块
设计理念：围绕任务全生命周期采集指标
"""
from prometheus_client import (
    Counter, Gauge, Histogram, Summary,
    generate_latest, CONTENT_TYPE_LATEST
)
from flask import Response
import time

# ============ 计数器 (Counter) ============
# 任务执行总次数（按任务名、状态标签区分）
task_executions_total = Counter(
    'scheduler_task_executions_total',
    'Total number of task executions',
    ['task_name', 'status']  # status: success / failed / retry
)

# 任务重试总次数
task_retries_total = Counter(
    'scheduler_task_retries_total',
    'Total number of task retries',
    ['task_name', 'retry_reason']
)

# 告警触发总次数
alerts_triggered_total = Counter(
    'scheduler_alerts_triggered_total',
    'Total number of alerts triggered',
    ['task_name', 'alert_type', 'channel']
)

# ============ 仪表盘 (Gauge) ============
# 当前正在运行的任务数
task_running_gauge = Gauge(
    'scheduler_tasks_running',
    'Number of tasks currently running',
    ['worker_id']
)

# 任务队列长度
task_queue_length = Gauge(
    'scheduler_task_queue_length',
    'Number of tasks waiting in queue',
    ['queue_name']
)

# Worker 存活状态
worker_health = Gauge(
    'scheduler_worker_health',
    'Worker health status (1=healthy, 0=unhealthy)',
    ['worker_id', 'worker_host']
)

# ============ 直方图 (Histogram) ============
# 任务执行耗时分布
task_duration_seconds = Histogram(
    'scheduler_task_duration_seconds',
    'Task execution duration in seconds',
    ['task_name'],
    buckets=(0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0)
)

# ============ 摘要 (Summary) ============
# 任务延迟（计划时间与实际执行时间的差值）
task_latency_seconds = Summary(
    'scheduler_task_latency_seconds',
    'Task scheduling latency in seconds',
    ['task_name']
)

# ============ 指标导出接口 ============
def metrics_endpoint():
    """供 Prometheus 抓取的 /metrics 端点"""
    return Response(generate_latest(), mimetype=CONTENT_TYPE_LATEST)

3. 告警模块 (alerting.py)

"""
告警模块：支持多级告警、多渠道通知、告警收敛
"""
import json
import smtplib
import requests
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Any
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

class AlertLevel(Enum):
    """告警级别"""
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"

class AlertType(Enum):
    """告警类型"""
    TASK_FAILED = "task_failed"
    TASK_TIMEOUT = "task_timeout"
    TASK_RETRY_EXCEEDED = "task_retry_exceeded"
    WORKER_DOWN = "worker_down"
    QUEUE_OVERFLOW = "queue_overflow"
    SCHEDULER_HA_FAIL = "scheduler_ha_fail"

@dataclass
class Alert:
    """告警实体"""
    task_name: str
    alert_type: AlertType
    level: AlertLevel
    message: str
    details: Dict[str, Any] = field(default_factory=dict)
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    fingerprint: str = ""  # 告警指纹，用于去重

    def __post_init__(self):
        self.fingerprint = f"{self.task_name}:{self.alert_type.value}"

class AlertChannel(Enum):
    """告警渠道"""
    WECHAT_WORK = "wechat_work"
    DINGTALK = "dingtalk"
    EMAIL = "email"
    WEBHOOK = "webhook"

class AlertManager:
    """
    告警管理器
    - 支持多渠道分发
    - 支持告警收敛（相同指纹5分钟内不重复通知）
    - 支持告警升级（连续失败N次后升级为CRITICAL）
    """

    def __init__(self, config: dict):
        self.config = config
        self._alert_cache: Dict[str, float] = {}
        self._failure_count: Dict[str, int] = {}
        self._cooldown_seconds = config.get('cooldown_seconds', 300)
        self._upgrade_threshold = config.get('upgrade_threshold', 3)

    def send(self, alert: Alert):
        """发送告警（带收敛逻辑）"""
        now = time.time()
        cache_key = alert.fingerprint

        last_time = self._alert_cache.get(cache_key, 0)
        if now - last_time < self._cooldown_seconds:
            logger.info(f"告警收敛：{cache_key} 在冷却期内，跳过")
            return

        failure_count = self._failure_count.get(cache_key, 0) + 1
        self._failure_count[cache_key] = failure_count

        if failure_count >= self._upgrade_threshold:
            alert.level = AlertLevel.CRITICAL
            alert.message = f"[升级告警] 连续失败{failure_count}次 - {alert.message}"

        self._alert_cache[cache_key] = now

        from app.metrics import alerts_triggered_total
        for channel in self.config.get('enabled_channels', []):
            alerts_triggered_total.labels(
                task_name=alert.task_name,
                alert_type=alert.alert_type.value,
                channel=channel
            ).inc()

        self._dispatch(alert)

    def _dispatch(self, alert: Alert):
        """分发告警到各渠道"""
        channels = self.config.get('enabled_channels', [])
        for channel in channels:
            try:
                if channel == AlertChannel.WECHAT_WORK.value:
                    self._send_wechat_work(alert)
                elif channel == AlertChannel.DINGTALK.value:
                    self._send_dingtalk(alert)
                elif channel == AlertChannel.EMAIL.value:
                    self._send_email(alert)
                elif channel == AlertChannel.WEBHOOK.value:
                    self._send_webhook(alert)
            except Exception as e:
                logger.error(f"告警发送失败 [{channel}]: {e}")

    def _send_wechat_work(self, alert: Alert):
        webhook_url = self.config.get('wechat_work', {}).get('webhook_url')
        if not webhook_url:
            return

        payload = {
            "msgtype": "markdown",
            "markdown": {
                "content": f"""
					## 🚨 任务告警 [{alert.level.value.upper()}]
					> **任务名称**: `{alert.task_name}`
					> **告警类型**: {alert.alert_type.value}
					> **告警时间**: {alert.timestamp}
					> **告警级别**: {alert.level.value}
					> **详细信息**: {alert.message}
					请及时处理！
                """.strip()
            }
        }
        requests.post(webhook_url, json=payload, timeout=10)

    def _send_dingtalk(self, alert: Alert):
        webhook_url = self.config.get('dingtalk', {}).get('webhook_url')
        if not webhook_url:
            return

        payload = {
            "msgtype": "markdown",
            "markdown": {
                "title": f"任务告警 - {alert.task_name}",
                "text": f"## ⚠️ 任务告警\n\n"
                        f"- **任务**: {alert.task_name}\n"
                        f"- **类型**: {alert.alert_type.value}\n"
                        f"- **级别**: {alert.level.value}\n"
                        f"- **时间**: {alert.timestamp}\n"
                        f"- **详情**: {alert.message}\n"
            }
        }
        requests.post(webhook_url, json=payload, timeout=10)

    def _send_email(self, alert: Alert):
        smtp_config = self.config.get('email', {})
        msg = MIMEMultipart()
        msg['From'] = smtp_config.get('sender')
        msg['To'] = ', '.join(smtp_config.get('recipients', []))
        msg['Subject'] = f"[任务告警-{alert.level.value}] {alert.task_name}"

        body = f"""
        <html><body>
            <h2 style="color: red;">🚨 任务执行异常告警</h2>
            <table border="1" cellpadding="8">
                <tr><td><b>任务名称</b></td><td>{alert.task_name}</td></tr>
                <tr><td><b>告警类型</b></td><td>{alert.alert_type.value}</td></tr>
                <tr><td><b>告警级别</b></td><td>{alert.level.value}</td></tr>
                <tr><td><b>告警时间</b></td><td>{alert.timestamp}</td></tr>
                <tr><td><b>详细信息</b></td><td>{alert.message}</td></tr>
            </table>
        </body></html>
        """
        msg.attach(MIMEText(body, 'html'))

        with smtplib.SMTP(smtp_config.get('host'), smtp_config.get('port', 25)) as server:
            if smtp_config.get('use_tls'):
                server.starttls()
            if smtp_config.get('username'):
                server.login(smtp_config['username'], smtp_config.get('password', ''))
            server.send_message(msg)

    def _send_webhook(self, alert: Alert):
        url = self.config.get('webhook', {}).get('url')
        if not url:
            return
        requests.post(url, json={
            'task_name': alert.task_name,
            'alert_type': alert.alert_type.value,
            'level': alert.level.value,
            'message': alert.message,
            'details': alert.details,
            'timestamp': alert.timestamp
        }, timeout=10)

    def clear_failure_count(self, task_name: str):
        keys_to_clear = [k for k in self._failure_count if task_name in k]
        for key in keys_to_clear:
            del self._failure_count[key]
        
        cache_keys_to_clear = [k for k in self._alert_cache if task_name in k]
        for key in cache_keys_to_clear:
            del self._alert_cache[key]

4. 调度引擎 + Worker 执行器 (scheduler.py)

"""
企业级调度引擎：基于 APScheduler + Celery 的混合架构
- 使用 APScheduler 做时间触发
- 使用 Celery 做分布式执行
- 集成 Prometheus 指标采集
- 集成失败告警
"""
import logging
import time
import traceback
from datetime import datetime
from functools import wraps
from typing import Callable

from celery import Celery, signals
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor

from app.metrics import (
    task_executions_total,
    task_retries_total,
    task_running_gauge,
    task_duration_seconds,
    task_latency_seconds,
    worker_health
)
from app.alerting import AlertManager, Alert, AlertType, AlertLevel

logger = logging.getLogger(__name__)

# ============ Celery 应用配置 ============
celery_app = Celery(
    'enterprise_scheduler',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',
)

celery_app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='Asia/Shanghai',
    enable_utc=True,
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    task_default_retry_delay=60,
    task_max_retries=3,
    task_soft_time_limit=300,
    task_time_limit=600,
    worker_prefetch_multiplier=1,
    worker_max_tasks_per_child=100,
    worker_concurrency=4,
)

# ============ 告警管理器实例 ============
alert_manager = AlertManager(config={
    'cooldown_seconds': 300,
    'upgrade_threshold': 3,
    'enabled_channels': ['wechat_work', 'email', 'webhook'],
    'wechat_work': {'webhook_url': 'https://qyapi.weixin.qq.com/...'},
    'email': {
        'host': 'smtp.example.com', 'port': 587, 'use_tls': True,
        'username': 'alert@example.com', 'password': 'pwd',
        'sender': 'alert@example.com', 'recipients': ['ops@example.com']
    },
    'webhook': {'url': 'https://your-system.example.com/api/alerts'}
})

# ============ 指标装饰器 ============
def track_task_metrics(task_name: str):
    def decorator(func: Callable):
        @wraps(func)
        def wrapper(*args, **kwargs):
            scheduled_time = kwargs.pop('_scheduled_time', None)
            if scheduled_time:
                latency = time.time() - scheduled_time
                task_latency_seconds.labels(task_name=task_name).observe(latency)

            task_running_gauge.labels(worker_id='worker-1').inc()
            start_time = time.time()
            status = 'success'

            try:
                return func(*args, **kwargs)
            except Exception as e:
                status = 'failed'
                alert_manager.send(Alert(
                    task_name=task_name,
                    alert_type=AlertType.TASK_FAILED,
                    level=AlertLevel.CRITICAL,
                    message=f"任务执行失败: {str(e)}\n{traceback.format_exc()}",
                    details={'args': str(args), 'kwargs': str(kwargs)}
                ))
                raise
            finally:
                duration = time.time() - start_time
                task_duration_seconds.labels(task_name=task_name).observe(duration)
                task_executions_total.labels(task_name=task_name, status=status).inc()
                task_running_gauge.labels(worker_id='worker-1').dec()

                if status == 'success':
                    alert_manager.clear_failure_count(task_name)

                logger.info(f"任务 [{task_name}] | 状态: {status} | 耗时: {duration:.2f}s")
        return wrapper
    return decorator

# ============ 任务定义示例 ============
@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
@track_task_metrics("data_sync_task")
def data_sync_task(self, source: str, target: str):
    logger.info(f"开始同步数据: {source} -> {target}")
    time.sleep(2)
    return {"status": "synced", "rows": 1000}

@celery_app.task(bind=True, max_retries=5, default_retry_delay=30)
@track_task_metrics("report_generation")
def report_generation(self, report_type: str, date: str):
    logger.info(f"生成报表: {report_type} for {date}")
    time.sleep(5)
    return {"report": report_type, "date": date}

# ============ APScheduler 调度配置 ============
def create_scheduler():
    scheduler = BackgroundScheduler()

    scheduler.add_jobstore(
        SQLAlchemyJobStore(url='sqlite:///scheduler_jobs.db'),
        alias='default'
    )
    scheduler.add_executor(ThreadPoolExecutor(max_workers=5), alias='thread')
    scheduler.add_executor(ProcessPoolExecutor(max_workers=2), alias='process')

    scheduler.add_job(
        data_sync_task.apply_async,
        trigger='interval',
        minutes=5,
        kwargs={'source': 'mysql', 'target': 'es'},
        id='data_sync_job',
        replace_existing=True,
        coalesce=True,
        max_instances=1
    )
    
    scheduler.add_job(
        report_generation.apply_async,
        trigger='cron',
        hour=2,
        minute=0,
        kwargs={'report_type': 'daily_sales', 'date': 'yesterday'},
        id='daily_report_job',
        replace_existing=True
    )
    return scheduler

# ============ Celery 信号处理 ============
@signals.task_failure.connect
def on_task_failure(sender, task_id, exception, args, kwargs, traceback, **extra):
    task_name = sender.name if sender else 'unknown'
    alert_manager.send(Alert(
        task_name=task_name,
        alert_type=AlertType.TASK_FAILED,
        level=AlertLevel.CRITICAL,
        message=str(exception),
        details={'task_id': task_id}
    ))

@signals.task_retry.connect
def on_task_retry(sender, task_id, reason, **extra):
    task_name = sender.name if sender else 'unknown'
    task_retries_total.labels(task_name=task_name, retry_reason=str(reason)).inc()

# ============ 主入口 ============
if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    scheduler = create_scheduler()
    scheduler.start()
    logger.info("🚀 调度引擎已启动")
    import threading
    threading.Event().wait()

5. Prometheus 配置 (prometheus.yml)

global:
  scrape_interval: 15s

rule_files:
  - 'alert_rules.yml'

scrape_configs:
  - job_name: 'scheduler'
    metrics_path: '/metrics'
    static_configs:
      - targets: ['scheduler-app:8000']

6. 告警规则 (alert_rules.yml)

groups:
  - name: scheduler_alerts
    rules:
      - alert: HighTaskFailureRate
        expr: |
          sum(rate(scheduler_task_executions_total{status="failed"}[5m]))
          /
          sum(rate(scheduler_task_executions_total[5m]))
          > 0.2
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "任务 {{ $labels.task_name }} 失败率过高"

      - alert: TaskSlowExecution
        expr: histogram_quantile(0.95, scheduler_task_duration_seconds_bucket) > 120
        for: 3m
        labels:
          severity: warning
        annotations:
          summary: "任务 {{ $labels.task_name }} P95 耗时超标"

---

五、关键设计模式与最佳实践

1. 分布式锁（防止重复执行）

import redis
import uuid

class DistributedLock:
    """基于 Redis 的分布式锁"""
    def __init__(self, redis_client, lock_key, ttl=60):
        self.redis = redis_client
        self.lock_key = f"lock:{lock_key}"
        self.ttl = ttl
        self.identifier = str(uuid.uuid4())

    def acquire(self):
        return self.redis.set(self.lock_key, self.identifier, nx=True, ex=self.ttl)

    def release(self):
        script = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        end
        return 0
        """
        self.redis.eval(script, 1, self.lock_key, self.identifier)

    def __enter__(self):
        while not self.acquire():
            time.sleep(1)
        return self

    def __exit__(self, *args):
        self.release()

2. 任务幂等性设计

import hashlib
import json

def idempotent():
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            r = redis.Redis()
            params_hash = hashlib.md5(
                json.dumps({'args': args, 'kwargs': kwargs}, sort_keys=True).encode()
            ).hexdigest()
            task_id = f"{func.__name__}:{params_hash}"

            if cached := r.get(f"idempotent:{task_id}"):
                return json.loads(cached)

            result = func(*args, **kwargs)
            r.setex(f"idempotent:{task_id}", 86400, json.dumps(result))
            return result
        return wrapper
    return decorator

---

六、最终总结

6.1 方案选型决策树

需要定时任务
├── 简单脚本/个人项目
│   └── schedule 库 / time.sleep
├── 单机中型项目
│   └── APScheduler（持久化 + 多线程/进程）
├── 分布式/微服务架构
│   ├── Celery Beat + Worker（Python生态首选）
│   ├── Airflow（数据流水线/DAG编排）
│   └── XXL-Job（Java生态，跨语言可调）
└── 超大规模/跨集群
    └── 自研调度平台（时间轮 + 一致性哈希 + 分片）

6.2 核心知识点总结

知识领域	关键要点
时间触发	cron 表达式、interval 轮询、时间轮算法
任务持久化	JobStore 抽象、SQLAlchemy/Redis 后端、幂等性
分布式调度	选主机制（ZooKeeper/ETCD）、一致性哈希分片
失败处理	指数退避重试、死信队列、熔断降级、告警收敛
可观测性	Prometheus 四大指标、RED 方法论
告警设计	多级告警、告警收敛/抑制、告警升级
高可用	无状态 Worker、Scheduler 主备切换、优雅停机

6.3 生产环境 Checklist

• ✅ 任务持久化（服务重启不丢失）
• ✅ 失败自动重试（指数退避 + 最大次数限制）
• ✅ 死信队列（超过重试次数的任务进入死信）
• ✅ 超时控制（软超时 + 硬超时）
• ✅ 幂等设计（防止重复执行副作用）
• ✅ 分布式锁（防止多节点重复执行）
• ✅ Prometheus 指标采集
• ✅ Grafana 可视化
• ✅ AlertManager 多级告警
• ✅ 优雅停机（SIGTERM 信号处理）
• ✅ 日志链路追踪（trace_id）

---

一句话总结：Python 定时任务从 time.sleep 到企业级平台，本质是解决 触发精度、执行可靠性、故障可观测性、规模可扩展性 四个核心问题。选对工具只是第一步，真正的挑战在于 失败处理、幂等设计、监控告警、优雅运维 这些工程细节。