python基础代码、优化、扩展和监控的完整示例。此示例使用 Celery
配合 RabbitMQ
作为消息代理,实现异步任务的调度、重试、定时任务以及错误监控等功能。
项目结构
我们将项目结构组织如下,以便代码逻辑清晰且易于扩展:
project/
│
├── celery_app.py # Celery应用的配置和初始化
├── tasks.py # 异步任务的定义
├── monitor.py # 异常监控和报警
└── main.py # 测试异步任务调用
1. celery_app.py
- 配置 Celery 应用
# celery_app.py
from celery import Celery
from celery.schedules import crontabapp = Celery('tasks', broker='amqp://localhost//', backend='redis://localhost')# 基础配置
app.conf.update(result_expires=3600, # 任务结果过期时间task_acks_late=True, # 确保任务执行后才确认完成worker_prefetch_multiplier=1, # 单次预取任务数task_serializer='json', # 任务数据序列化格式result_serializer='json', # 任务结果序列化格式accept_content=['json'], # 仅接收json格式task_soft_time_limit=300, # 软超时时间task_time_limit=600, # 硬超时时间worker_hijack_root_logger=False, # 不劫持主日志worker_log_format='[%(asctime)s: %(levelname)s/%(processName)s] %(message)s',
)# 定时任务配置
app.conf.beat_schedule = {'scheduled_add': {'task': 'tasks.add','schedule': crontab(hour=7, minute=30, day_of_week=1),'args': (16, 16),},
}# 任务路由配置:不同的任务可以走不同的队列
app.conf.task_routes = {'tasks.add': {'queue': 'high_priority'},
}
2. tasks.py
- 定义任务
# tasks.py
from celery_app import app
from monitor import task_failure_handler
import time# 定义基础任务
@app.task(bind=True, max_retries=3)
def add(self, x, y):try:time.sleep(5) # 模拟耗时任务return x + yexcept Exception as exc:raise self.retry(exc=exc, countdown=5) # 5秒后重试
3. monitor.py
- 监控与报警
# monitor.py
from celery.signals import task_failure@task_failure.connect
def task_failure_handler(sender=None, exception=None, **kwargs):# 发送报警通知或记录错误日志print(f"[ALERT] Task {sender.name} failed due to {exception}")
4. main.py
- 测试任务调用
# main.py
from tasks import add
from celery_app import appif __name__ == "__main__":# 启动异步任务result = add.delay(4, 6)print("Task state:", result.state) # 打印任务状态print("Result:", result.get()) # 获取任务结果(阻塞等待)# 组合任务示例:Groupfrom celery import groupgroup_tasks = group(add.s(i, i) for i in range(10))group_result = group_tasks.apply_async()print("Group Result:", group_result.get())# 链式任务示例:Chordfrom celery import chordcallback = add.s(10, 20)chord_tasks = chord((add.s(i, i) for i in range(10)), callback)chord_result = chord_tasks.apply_async()print("Chord Result:", chord_result.get())
运行和监控
-
启动 RabbitMQ 服务:
sudo service rabbitmq-server start
-
启动 Celery Worker:
运行以下命令,指定high_priority
队列处理高优先级任务。celery -A celery_app worker -Q high_priority,default -l info
-
启动 Celery Beat(用于调度定时任务):
celery -A celery_app beat -l info
-
启动 Flower 实时监控(可选):
celery -A celery_app flower --port=5555
访问
http://localhost:5555
进行任务和 worker 状态的实时监控。 -
Prometheus 和 Grafana 监控(可选):
配置 Celery 的自定义事件,并使用 Prometheus 采集数据,再通过 Grafana 可视化 Celery 的性能指标。
此示例项目具有以下特性:
- 异步任务:通过
delay()
方法调用。 - 重试机制:在任务异常时自动重试。
- 任务调度:支持定时任务,利用
Celery Beat
实现周期性任务调度。 - 报警机制:在任务失败时发送报警或日志记录。
- 监控系统:使用
Flower
进行实时监控,支持 Prometheus 和 Grafana 扩展。