Python 需要异步执行后台任务…

在 Python 项目中用 Celery 将耗时任务异步化、定时化和分布式执行，支持重试、工作流和多种框架集成。

适用场景

解决思路

1. 安装基础 Celery：pip install celery
2. 安装 Redis broker 支持：pip install celery[redis]
3. 安装 RabbitMQ broker 支持：pip install celery[amqp]
4. 安装完整依赖：pip install celery[redis,msgpack,auth,cassandra,elasticsearch,s3,sqs]
5. 创建 Celery app：app = Celery('myapp', broker='redis://localhost:6379/0', backend='redis://localhost:6379/1')
6. 配置序列化与时区：app.conf.update(task_serializer='json', accept_content=['json'], result_serializer='json', timezone='UTC', enable_utc=True)
7. 用 @app.task 装饰器定义任务，如 def add(x, y): return x + y
8. 启动 worker：celery -A celery_app worker --loglevel=info
9. 启动多并发 worker：celery -A celery_app worker --concurrency=4 --loglevel=info
10. 启动指定队列的命名 worker：celery -A celery_app worker -Q emails,reports --loglevel=info
11. 异步调用任务：result = add.delay(4, 6)
12. 等待结果：result.get(timeout=10)
13. 使用 apply_async 延迟执行：send_email.apply_async(args=['user@example.com', 'Hello', 'Welcome!'], countdown=60)
14. 检查任务状态：result.status
15. 配置 Redis broker 和 backend：broker_url = 'redis://localhost:6379/0'，result_backend = 'redis://localhost:6379/1'
16. Redis 认证：broker_url = 'redis://:password@localhost:6379/0'
17. 配置 Redis Sentinel：broker_url = 'sentinel://localhost:26379;sentinel://localhost:26380'，并设置 broker_transport_options 的 master_name 和 sentinel_kwargs
18. Redis 连接池设置：broker_pool_limit = 10，broker_connection_retry = True，broker_connection_retry_on_startup = True，broker_connection_max_retries = 10
19. 配置 RabbitMQ：broker_url = 'amqp://guest:guest@localhost:5672//'
20. RabbitMQ 虚拟主机：broker_url = 'amqp://user:password@localhost:5672/myvhost'
21. RabbitMQ 高可用多 broker：broker_url = ['amqp://user:password@host1:5672//', 'amqp://user:password@host2:5672//']
22. RabbitMQ 特定设置：broker_heartbeat = 30，broker_pool_limit = 10
23. 配置 Amazon SQS：broker_url = 'sqs://'，并设置 broker_transport_options 的 region、queue_name_prefix、visibility_timeout、polling_interval
24. SQS 自定义凭证与预定义队列：使用 boto3 和 broker_transport_options 的 predefined_queues
25. 任务定义方式1：使用 @app.task 装饰器
26. 任务定义方式2：使用 @shared_task 使任务与框架无关
27. 任务定义方式3：继承 Task 类并实现 on_success、on_failure、on_retry
28. 配置任务选项：name、bind=True、ignore_result=True、max_retries、default_retry_delay、rate_limit、time_limit、soft_time_limit、serializer、compression、priority、queue、routing_key、acks_late、reject_on_worker_lost
29. 使用 bind=True 访问任务上下文：self.request.id、self.name、self.request.args、self.request.kwargs、self.request.retries、self.request.delivery_info
30. 手动重试：raise self.retry(exc=exc, countdown=60, max_retries=3)
31. 使用 delay() 简单异步执行
32. 使用 apply_async() 完整控制：参数 args、kwargs、countdown、eta、expires、queue、routing_key、exchange、priority、serializer、compression、retry、retry_policy、link、link_error
33. 检查结果：result.ready()、result.get()、result.result、result.status、result.successful()、result.failed()、result.traceback、result.info
34. 释放结果：result.forget()
35. 撤销任务：result.revoke(terminate=True) 或 add.AsyncResult(task_id).revoke()
36. 创建任务签名：add.signature((2, 2), countdown=10) 或 add.s(2, 2)
37. 部分参数：partial = add.s(2)，然后 partial.apply_async(args=(4,))
38. 不可变签名：add.si(2, 2)
39. 克隆并修改签名：sig.clone(countdown=60)
40. 执行签名：sig.delay()、sig.apply_async()、sig() 同步执行
41. 定义队列：使用 kombu.Queue 和 Exchange，设置 app.conf.task_queues
42. 设置默认队列：app.conf.task_default_queue、task_default_exchange、task_default_routing_key
43. 配置任务路由规则：app.conf.task_routes 字典，如 {'myapp.tasks.send_email': {'queue': 'emails'}}
44. 使用函数路由：定义 route_task 函数并设置 app.conf.task_routes = (route_task,)
45. 启动 worker 消费指定队列：celery -A myapp worker -Q emails,reports --loglevel=info
46. 为不同队列启动多个 worker：celery -A myapp worker -Q high_priority -c 4，celery -A myapp worker -Q default -c 2，celery -A myapp worker -Q low_priority -c 1
47. 配置优先级队列：app.conf.task_queue_max_priority = 10，app.conf.task_default_priority = 5
48. 发送带优先级的任务：task.apply_async(args=(), priority=9)
49. 配置周期性任务：使用 beat_schedule 字典，设置 task、schedule、args、kwargs
50. 使用 crontab 调度：crontab(hour=7, minute=30)、crontab(day_of_week=1) 等
51. 使用 solar 调度：solar('sunrise', -37.81, 144.96)
52. 启动 beat 调度器：celery -A myapp beat --loglevel=info
53. 使用自定义调度器：celery -A myapp beat --scheduler django_celery_beat.schedulers:DatabaseScheduler
54. 开发时合并 worker 和 beat：celery -A myapp worker --beat --loglevel=info
55. 安装 django-celery-beat：pip install django-celery-beat
56. 配置 Django 的 INSTALLED_APPS 加入 django_celery_beat
57. 运行迁移：python manage.py migrate django_celery_beat
58. 创建 IntervalSchedule 和 PeriodicTask：使用 Django ORM 创建，如 IntervalSchedule.objects.get_or_create(every=10, period=IntervalSchedule.SECONDS)，PeriodicTask.objects.create(interval=schedule, name='Import feed every 10 seconds', task='myapp.tasks.import_feed', args=json.dumps(['https://example.com/feed']))
59. 创建 CrontabSchedule 和 PeriodicTask：CrontabSchedule.objects.get_or_create(minute='0', hour='*/4', day_of_week='*', day_of_month='*', month_of_year='*')，PeriodicTask.objects.create(crontab=schedule, name='Hourly cleanup', task='myapp.tasks.cleanup')
60. 使用 chain 顺序执行：chain(add.s(2, 2), add.s(4), add.s(8))() 或 (add.s(2, 2) | add.s(4) | add.s(8))()
61. 使用 group 并行执行：group([add.s(2, 2), add.s(4, 4), add.s(8, 8)]).apply_async()
62. 使用 chord 回调：chord([fetch_url.s(url) for url in urls])(combine_results.s())
63. 使用 map：add.map([(2, 2), (4, 4), (8, 8)])
64. 使用 starmap：add.starmap([(2, 2), (4, 4), (8, 8)])
65. 构建复杂工作流：组合 chain、group、chord，如 chain(fetch_data.s(source), group([process_chunk.s(chunk_id) for chunk_id in range(10)]), aggregate.s(), save_results.s())
66. 自动重试配置：@app.task(autoretry_for=(RequestException, IOError), retry_kwargs={'max_retries': 5}, retry_backoff=True, retry_backoff_max=600, retry_jitter=True)
67. 手动重试并分类异常：在任务中 try/except，对临时错误 raise self.retry(exc=exc, countdown=60)，对永久错误直接 raise
68. 使用错误回调：risky_task.apply_async(args=(data,), link_error=on_error.s())
69. 自定义 Task 类处理失败：继承 Task 实现 on_failure、on_success、on_retry，并 @app.task(base=CallbackTask)
70. 启用任务事件：app.conf.worker_send_task_events = True，app.conf.task_send_sent_event = True
71. 使用信号监听任务：@signals.task_prerun.connect、task_postrun、task_failure、task_retry
72. 安装并启动 Flower：pip install flower，celery -A myapp flower --port=5555
73. 配置 Flower：flower_basic_auth = ['admin:password']，flower_persistent = True，flower_db = 'flower.db'，flower_max_tasks = 10000
74. 使用 inspect 检查 worker：i = app.control.inspect()，i.active()、i.scheduled()、i.reserved()、i.stats()、i.registered()
75. 通过控制接口撤销任务：app.control.revoke(task_id, terminate=True)
76. 关闭 worker：app.control.shutdown()
77. 重启 worker 池：app.control.pool_restart()
78. 动态设置速率限制：app.control.rate_limit('myapp.tasks.slow_task', '10/m')
79. 命令行检查：celery -A myapp inspect active、celery -A myapp inspect scheduled、celery -A myapp inspect stats、celery -A myapp inspect registered
80. 命令行撤销任务：celery -A myapp control revoke <task_id>
81. 命令行关闭 worker：celery -A myapp control shutdown
82. 命令行清除任务：celery -A myapp purge
83. 自定义指标：在任务中使用 prometheus_client 的 Counter 和 Histogram
84. Django 集成：创建 myproject/celery.py，设置 DJANGO_SETTINGS_MODULE，app.config_from_object('django.conf:settings', namespace='CELERY')，app.autodiscover_tasks()
85. Django __init__.py 中导入 celery_app：from .celery import app as celery_app; __all__ = ('celery_app',)
86. Django settings 配置：CELERY_BROKER_URL、CELERY_RESULT_BACKEND、CELERY_TASK_SERIALIZER、CELERY_RESULT_SERIALIZER、CELERY_ACCEPT_CONTENT、CELERY_TIMEZONE、CELERY_ENABLE_UTC
87. Django 任务使用 @shared_task，在视图中调用 task.delay(...)
88. FastAPI 集成：创建 celery_app = Celery(...)，定义任务，在端点中调用 task = process_data.delay(data)，返回 task_id
89. FastAPI 状态查询：使用 AsyncResult(task_id, app=celery_app) 获取 status 和 result
90. Flask 集成：定义 make_celery(app)，创建 ContextTask 使任务在 app context 中运行，celery.Task = ContextTask
91. Flask 任务中使用 with app.app_context():
92. 测试使用 eager 模式：在 conftest.py 中设置 task_always_eager=True，task_eager_propagates=True
93. 使用真实 broker 测试：设置 broker_url 和 result_backend 为测试数据库，使用 celery_worker fixture 启动 worker
94. 模拟外部依赖：使用 unittest.mock.patch 修补 requests.get 等
95. 测试周期性任务：检查 app.conf.beat_schedule，直接调用任务函数
96. 集成测试工作流：使用 chain 并调用 result.get(timeout=30)
97. 生产 worker 配置：celery -A myapp worker --autoscale=10,3 --max-tasks-per-child=1000 --time-limit=300 --soft-time-limit=240 --loglevel=info --logfile=/var/log/celery/worker.log --pidfile=/var/run/celery/worker.pid
98. 使用 celery multi 启动多个专用 worker：celery multi start worker1 -A myapp -Q high_priority -c 4 --max-tasks-per-child=100 worker2 -A myapp -Q default -c 2 --max-tasks-per-child=1000 worker3 -A myapp -Q low_priority -c 1 --autoscale=3,1
99. 优雅停止：celery multi stop worker1 worker2 worker3 或 celery multi stopwait worker1 worker2 worker3
100. 生产配置最佳实践：设置 broker_url、result_backend、broker_connection_retry_on_startup=True、broker_pool_limit=50、result_expires=3600、序列化、时区、worker_prefetch_multiplier=4、worker_max_tasks_per_child=1000、task_acks_late=True、task_reject_on_worker_lost=True、task_track_started=True、task_time_limit=300、task_soft_time_limit=240、日志格式
101. systemd 服务：创建 celery.service 和 celerybeat.service，配置 ExecStart 使用 celery multi start 或 celery beat
102. Sentry 集成：pip install sentry-sdk，初始化 sentry_sdk.init(dsn='https://your-sentry-dsn', integrations=[CeleryIntegration()], traces_sample_rate=0.1)
103. 速率限制：设置 task_default_rate_limit = '100/m'，任务装饰器 rate_limit='10/m'，动态 app.control.rate_limit('myapp.tasks.slow_task', '5/m')
104. 健康检查：使用 app.control.inspect().stats() 和 app.control.ping(timeout=1.0) 检查 worker
105. 任务优化：对不需要结果的任务使用 ignore_result=True，大负载用 compression='gzip'，用 serializer='msgpack'
106. Worker 调优：根据 CPU/IO 密集型设置 worker_concurrency，调整 worker_prefetch_multiplier，设置 task_acks_late，worker_max_tasks_per_child，worker_max_memory_per_child
107. 数据库结果后端优化：使用 Redis 作为结果后端，或数据库配置 database_engine_options 和 result_expires=3600
108. 任务分块：将大列表分块，每块一个任务，使用 group(process_batch.s(chunk) for chunk in chunks(large_list, 100))
109. 连接池：设置 broker_pool_limit，redis_max_connections，SQLAlchemy 使用 QueuePool
110. 邮件发送任务：定义 send_email_task 带 bind=True, max_retries=3，捕获 SMTPException 并 self.retry，批量使用 rate_limit='100/m'
111. 报告生成任务：使用 self.update_state(state='PROGRESS', meta={'current': 0, 'total': 100}) 更新进度，生成 PDF，上传 S3，发送通知
112. 数据管道：用 chain 组合 fetch_data、clean_data、transform_data、load_data
113. Webhook 处理：任务中验证签名，根据事件类型处理，使用 autoretry_for=(RequestException,), max_retries=5，FastAPI 端点调用 process_webhook.delay(data)
114. 图片处理：定义 resize_image 任务，使用 chord([resize_image.s(image_path, size) for size in sizes])(upload_to_cdn.s())
115. 最佳实践：确保任务幂等，例如检查订单状态；任务小而专注；避免传数据库对象，传 ID；设置 time_limit 和 soft_time_limit
116. 错误处理最佳实践：分类异常，指数退避，设置最大重试，记录失败
117. 性能最佳实践：使用 ignore_result，批处理，优化序列化，连接池，任务分块
118. 监控最佳实践：启用事件，使用 Flower，健康检查，Sentry 集成
119. 安全最佳实践：验证输入，使用认证和加密的 broker，限制执行时间，速率限制
120. 故障排查：任务不执行时检查 worker 是否运行，启动 worker，检查队列和路由，检查 broker 连通性
121. 任务静默失败：启用 task_track_started=True，检查 result.traceback，查看 worker 日志，测试时用 eager 模式
122. 内存泄漏：设置 worker_max_tasks_per_child=1000，worker_max_memory_per_child=200000
123. 任务执行慢：添加计时日志，增加 worker 并发，优化任务代码，任务分块，增加 worker 数量
124. Broker 连接问题：测试 app.connection().connect()，检查 broker 运行状态，验证 URL，检查网络，启用 broker_connection_retry_on_startup=True
125. 结果不持久化：检查结果后端配置，任务是否 ignore_result=True，result_expires，测试后端连接
126. Beat 不调度：检查 beat 进程，验证 beat_schedule，查看 beat 日志，使用数据库调度器
127. Worker 崩溃：检查日志，设置 worker_max_tasks_per_child，添加时间限制，使用 systemd 自动重启，用 Flower 监控
128. 任务队列积压：增加 worker，提高并发，优化慢任务，使用任务路由，检查阻塞
129. 高级配置：自定义 Task 类管理数据库连接，在 after_return 关闭连接
130. 自定义序列化器：使用 kombu.serialization.register 注册自定义编码解码，设置 app.conf.task_serializer
131. 任务继承：创建 BaseTask 实现 on_failure 和 on_retry，用 @app.task(base=BaseTask) 应用

适用边界 · 注意事项

• 简单异步 I/O 场景，直接用 asyncio
• 实时请求/响应且要求亚秒级延迟，用内存队列或异步 Web 框架
• 基础设施极简且只需简单后台任务，考虑 RQ 或 Huey