使用 `Celery` 配合 `RabbitMQ` 作为消息代理，实现异步任务的调度、重试、定时任务以及错误监控等功能

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 crontab

app = 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 + y
    except 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 app

if __name__ == "__main__":
    # 启动异步任务
    result = add.delay(4, 6)
    print("Task state:", result.state)  # 打印任务状态
    print("Result:", result.get())      # 获取任务结果（阻塞等待）
    
    # 组合任务示例：Group
    from celery import group
    group_tasks = group(add.s(i, i) for i in range(10))
    group_result = group_tasks.apply_async()
    print("Group Result:", group_result.get())
    
    # 链式任务示例：Chord
    from celery import chord
    callback = 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())

---

运行和监控

1. 启动 RabbitMQ 服务：

   sudo service rabbitmq-server start
   

2. 启动 Celery Worker：
   运行以下命令，指定 high_priority 队列处理高优先级任务。

   celery -A celery_app worker -Q high_priority,default -l info
   

3. 启动 Celery Beat（用于调度定时任务）：

   celery -A celery_app beat -l info
   

4. 启动 Flower 实时监控（可选）：

   celery -A celery_app flower --port=5555
   

   访问 http://localhost:5555 进行任务和 worker 状态的实时监控。

5. Prometheus 和 Grafana 监控（可选）：
   配置 Celery 的自定义事件，并使用 Prometheus 采集数据，再通过 Grafana 可视化 Celery 的性能指标。

---

此示例项目具有以下特性：

• 异步任务：通过 delay() 方法调用。
• 重试机制：在任务异常时自动重试。
• 任务调度：支持定时任务，利用 Celery Beat 实现周期性任务调度。
• 报警机制：在任务失败时发送报警或日志记录。
• 监控系统：使用 Flower 进行实时监控，支持 Prometheus 和 Grafana 扩展。

原文地址：https://blog.csdn.net/u014158430/article/details/143489303

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！