需要设计或优化 Celery 分布式任务队列、异步后台作业…

适用于用 Python Celery 处理异步任务、后台作业、链/组/和弦工作流、Beat 调度和生产监控的场景。

适用场景

解决思路

1. 先用 TDD：在 tests/test_tasks.py 写失败测试，配置 celery_config 为 broker_url='memory://'、result_backend='cache+memory://'、task_always_eager=True、task_eager_propagates=True。
2. 用 celery_app、celery_worker fixture 测试 process_order：调用 process_order.delay(order_id=123)，断言 result.get(timeout=10) == {'order_id': 123, 'status': 'success'}。
3. 测试幂等：连续两次 process_order.delay(order_id=123).get(timeout=10)，断言两次 status 都属于 ['success', 'already_processed']。
4. 测试重试：用 mocker.patch 让 perform_order_processing 第一次抛 TemporaryError('Timeout')、第二次返回 {'result': 'ok'}，断言最终 status=='success' 且 mock_process.call_count == 2。
5. 实现最小可过代码：在 myapp/tasks.py 创建 Celery('tasks', broker='redis://localhost:6379/0')，用 @app.task(bind=True, max_retries=3) 定义 process_order。
6. 在 process_order 中：获取 order；若 order.status == 'processed' 返回 {'order_id': order_id, 'status': 'already_processed'}；否则调用 perform_order_processing(order)，返回 {'order_id': order_id, 'status': 'success'}。
7. 在 process_order 捕获 TemporaryError 时 raise self.retry(exc=exc, countdown=2 ** self.request.retries)。
8. 重构时添加 proper error handling、time limits、observability。
9. 运行验证：pytest tests/test_tasks.py -v。
10. 运行覆盖率：pytest tests/test_tasks.py --cov=myapp.tasks --cov-report=term-missing。
11. 运行工作流测试：pytest tests/test_workflows.py -v。
12. 运行真实 broker 集成测试：pytest tests/integration/ --broker=redis://localhost:6379/0。
13. 性能模式-任务分块：不要 for item_id in item_ids: process_item.delay(item_id)，而是定义 process_batch(item_ids: list)，对 chunks(item_ids, size=100) 批量 fetch_items_bulk(chunk) 并 process(item)，再按 chunk 调用 process_batch.delay(chunk)。
14. 预取调优 CPU-bound：设置 app.conf.worker_prefetch_multiplier = 4，运行 celery -A app worker --concurrency=4。
15. 预取调优 I/O-bound：设置 app.conf.worker_prefetch_multiplier = 1，运行 celery -A app worker --pool=gevent --concurrency=100。
16. 长任务：设置 app.conf.worker_prefetch_multiplier = 1 和 app.conf.task_acks_late = True。
17. 结果后端优化 fire-and-forget：对 send_email 这类任务使用 @app.task(ignore_result=True)，不返回无意义结果。
18. 需要结果时设置 app.conf.result_expires = 3600。
19. 大结果：将大数据 save_to_s3(data) 后只返回 {'result_key': result_key}。
20. 连接池：模块级 create_engine('postgresql://user:pass@localhost/db', pool_size=20, max_overflow=10, pool_pre_ping=True)，并创建 redis_pool = ConnectionPool(host='localhost', port=6379, max_connections=50)。
21. 任务中数据库使用 with db_engine.connect() as conn: return conn.execute(query).fetchall()；Redis 使用 Redis(connection_pool=redis_pool) 并 set(key, value)。
22. 任务路由：用 Queue/Exchange 定义 critical、default、bulk 队列，配置 app.conf.task_routes 将 tasks.critical_payment 路由到 critical，将 tasks.generate_report 路由到 bulk。
23. 按队列启动专用 worker：celery -A app worker -Q critical --concurrency=4；celery -A app worker -Q bulk --concurrency=2。
24. 任务设计与工作流：用 @app.task 或 @shared_task 定义任务；实现幂等任务；用 chain 顺序执行、group 并行、chord map-reduce；将任务路由到指定队列/worker；避免长任务，拆成子任务。
25. Broker 配置：Redis 适合简单场景，RabbitMQ 适合可靠性；配置连接池、heartbeats、failover；启用 broker 认证和 TLS 加密；监控 broker 健康与连接状态。
26. 任务可靠性：实现带 exponential backoff 的重试；关键任务用 acks_late=True；设置 soft/hard time limits；用 error callbacks 处理异常；实现 dead letter queues；设计幂等任务。
27. 结果后端：选择 Redis、database、RPC 等合适后端；设置 result expiration 防内存泄漏；fire-and-forget 用 ignore_result=True；结果只存最小数据，大数据用外部存储。
28. Celery Beat：用 crontab 定义周期任务，用 interval 定义简单周期任务，配置 Beat scheduler persistence（database backend），用 task locks 避免调度冲突。
29. 监控与可观测性：部署 Flower 实时监控；导出 Prometheus metrics；跟踪 task success/failure rates 和 queue lengths；用 correlation IDs 实现分布式追踪；带上下文日志记录任务执行。
30. 完整任务定义：@app.task(bind=True, name='tasks.process_order', max_retries=3, default_retry_delay=60, acks_late=True, reject_on_worker_lost=True, time_limit=300, soft_time_limit=240, rate_limit='100/m')。
31. 任务内记录 logger.info(f'Processing order {order_id}', extra={'task_id': self.request.id})；处理订单；返回 status。
32. 任务异常：捕获 SoftTimeLimitExceeded 时 cleanup_processing(order_id) 后 raise；捕获 TemporaryError 时 self.retry(exc=exc, countdown=2 ** self.request.retries)；捕获 PermanentError 时 send_failure_notification(order_id, str(exc)) 后 raise。
33. 工作流 chain：chain(fetch_data.s('https://api.example.com/data'), process_item.s(), send_notification.s())。
34. 工作流 group：group(fetch_data.s(url) for url in urls)。
35. 工作流 chord：chord(group(process_item.s(item) for item in items))(aggregate_results.s())。
36. 生产配置：app.conf.update 设置 broker_url='redis://localhost:6379/0'、broker_connection_retry_on_startup=True、broker_pool_limit=10、result_backend='redis://localhost:6379/1'、result_expires=3600。
37. 生产配置序列化：task_serializer='json'、result_serializer='json'、accept_content=['json']。
38. 生产配置可靠性：task_acks_late=True、task_reject_on_worker_lost=True、task_time_limit=300、task_soft_time_limit=240。
39. 生产配置 worker：worker_prefetch_multiplier=4、worker_max_tasks_per_child=1000。
40. 重试策略：@app.task(bind=True, max_retries=5, autoretry_for=(RequestException,), retry_backoff=True, retry_backoff_max=600, retry_jitter=True) 定义 call_external_api(url: str)。
41. call_external_api 中 requests.get(url, timeout=10)，response.raise_for_status()，返回 response.json()。
42. Beat 调度配置：app.conf.beat_schedule 定义 'cleanup-temp-files' 使用 timedelta(minutes=10)，'daily-report' 使用 crontab(hour=3, minute=0)。
43. 安全序列化：绝不 app.conf.task_serializer = 'pickle'；使用 task_serializer='json'、result_serializer='json'、accept_content=['json']。
44. Broker 认证与 TLS：Redis 使用 broker_url='redis://:password@localhost:6379/0'，broker_use_ssl={'ssl_cert_reqs': 'required', 'ssl_ca_certs': '/path/to/ca.pem'}；RabbitMQ 使用 broker_url='amqps://user:password@localhost:5671/vhost'。
45. 输入验证：用 pydantic BaseModel 定义 OrderData(order_id: int, amount: float)；在 process_order_validated(order_data: dict) 中 validated = OrderData(**order_data)，调用 process_order(validated.dict())。
46. 避免使用 pickle 序列化；改用 json。
47. 避免非幂等任务：不要 increment_counter；改为 set_counter(user_id, value)。
48. 避免缺少时间限制：@app.task(time_limit=30, soft_time_limit=25) 包裹 external_api_call。
49. 避免存大结果：process_file 不要 return read_large_file(file_id)；改为 result_id = save_to_storage(read_large_file(file_id))，返回 {'result_id': result_id}。
50. 预实现 Phase 1：写失败测试；定义任务幂等策略；选择队列路由；确定结果存储需求（ignore_result?）；规划重试与错误处理；审查安全需求（序列化、认证）。
51. 预实现 Phase 2：任务设置 soft/hard time limits；关键任务使用 acks_late=True；用 Pydantic 验证输入；用 correlation ID 记录日志；为 DB/Redis 配置连接池；大结果外部存储。
52. 提交前 Phase 3：pytest tests/test_tasks.py -v 全部通过；pytest --cov=myapp.tasks 覆盖率足够；序列化为 JSON；配置 broker 认证；设置 result expiration；配置 Flower/Prometheus；记录 task routes；实现 dead letter queue handling。
53. NEVER：使用 pickle 序列化；无时间限制运行；在结果中存大对象；创建非幂等任务；无 broker 认证运行；无认证暴露 Flower。
54. ALWAYS：使用 JSON 序列化；设置 soft/hard time limits；任务幂等；关键任务 acks_late=True；设置 result expiration；实现带 backoff 的重试；用 Flower/Prometheus 监控；验证任务输入；用 correlation ID 记录日志。

适用边界 · 注意事项

• 不使用 Python/Celery 或不需要分布式任务队列的简单同步场景。
• 无法部署 Redis/RabbitMQ 等 broker、worker 或 Flower 监控的环境。
• 只需要轻量定时脚本、无需重试/路由/结果后端/可观测性的场景。