利用 `Celery` 共同 `RabbitMQ` 作为消息代理，实现异步使命的调度、重试、 ...

python基础代码、优化、扩展和监控的完整示例。此示例利用 Celery 共同 RabbitMQ 作为消息代理，实现异步使命的调度、重试、定时使命以及错误监控等功能。

---

项目结构

我们将项目结构组织如下，以便代码逻辑清楚且易于扩展：

1. project/
2. │
3. ├── celery_app.py        # Celery应用的配置和初始化
4. ├── tasks.py             # 异步任务的定义
5. ├── monitor.py           # 异常监控和报警
6. └── main.py              # 测试异步任务调用

复制代码

1. celery_app.py - 设置 Celery 应用

1. # celery_app.py
2. from celery import Celery
3. from celery.schedules import crontab
5. app = Celery('tasks', broker='amqp://localhost//', backend='redis://localhost')
7. # 基础配置
8. app.conf.update(
9.     result_expires=3600,                     # 任务结果过期时间
10.     task_acks_late=True,                     # 确保任务执行后才确认完成
11.     worker_prefetch_multiplier=1,            # 单次预取任务数
12.     task_serializer='json',                  # 任务数据序列化格式
13.     result_serializer='json',                # 任务结果序列化格式
14.     accept_content=['json'],                 # 仅接收json格式
15.     task_soft_time_limit=300,                # 软超时时间
16.     task_time_limit=600,                     # 硬超时时间
17.     worker_hijack_root_logger=False,         # 不劫持主日志
18.     worker_log_format='[%(asctime)s: %(levelname)s/%(processName)s] %(message)s',
19. )
21. # 定时任务配置
22. app.conf.beat_schedule = {
23.     'scheduled_add': {
24.         'task': 'tasks.add',
25.         'schedule': crontab(hour=7, minute=30, day_of_week=1),
26.         'args': (16, 16),
27.     },
28. }
30. # 任务路由配置：不同的任务可以走不同的队列
31. app.conf.task_routes = {
32.     'tasks.add': {'queue': 'high_priority'},
33. }

复制代码

2. tasks.py - 定义使命

1. # tasks.py
2. from celery_app import app
3. from monitor import task_failure_handler
4. import time
6. # 定义基础任务
7. @app.task(bind=True, max_retries=3)
8. def add(self, x, y):
9.     try:
10.         time.sleep(5)  # 模拟耗时任务
11.         return x + y
12.     except Exception as exc:
13.         raise self.retry(exc=exc, countdown=5)  # 5秒后重试

复制代码

3. monitor.py - 监控与报警

1. # monitor.py
2. from celery.signals import task_failure
4. @task_failure.connect
5. def task_failure_handler(sender=None, exception=None, **kwargs):
6.     # 发送报警通知或记录错误日志
7.     print(f"[ALERT] Task {sender.name} failed due to {exception}")

复制代码

4. main.py - 测试使命调用

1. # main.py
2. from tasks import add
3. from celery_app import app
5. if __name__ == "__main__":
6.     # 启动异步任务
7.     result = add.delay(4, 6)
8.     print("Task state:", result.state)  # 打印任务状态
9.     print("Result:", result.get())      # 获取任务结果（阻塞等待）
10.    
11.     # 组合任务示例：Group
12.     from celery import group
13.     group_tasks = group(add.s(i, i) for i in range(10))
14.     group_result = group_tasks.apply_async()
15.     print("Group Result:", group_result.get())
16.    
17.     # 链式任务示例：Chord
18.     from celery import chord
19.     callback = add.s(10, 20)
20.     chord_tasks = chord((add.s(i, i) for i in range(10)), callback)
21.     chord_result = chord_tasks.apply_async()
22.     print("Chord Result:", chord_result.get())

复制代码

---

运行和监控

• 启动 RabbitMQ 服务：
  
  1. sudo service rabbitmq-server start

  复制代码
• 启动 Celery Worker：
  运行以下命令，指定 high_priority 队列处理高优先级使命。
  
  1. celery -A celery_app worker -Q high_priority,default -l info

  复制代码
• 启动 Celery Beat（用于调度定时使命）：
  
  1. celery -A celery_app beat -l info

  复制代码
• 启动 Flower 及时监控（可选）：
  
  1. celery -A celery_app flower --port=5555

  复制代码
  访问 http://localhost:5555 进行使命和 worker 状态的及时监控。
  
• Prometheus 和 Grafana 监控（可选）：
  设置 Celery 的自定义事件，并利用 Prometheus 采集数据，再通过 Grafana 可视化 Celery 的性能指标。
  

---

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

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

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。