background-job-processing

后台作业处理

目录

• 概述
• 适用场景
• 快速开始
• 参考指南
• 最佳实践

概述

构建健壮的后台作业处理系统，使用分布式任务队列、工作池、作业调度、错误处理、重试策略和监控，高效执行异步任务。

适用场景

• 异步处理长时间运行的操作
• 在后台发送电子邮件
• 生成报告或导出文件
• 处理大型数据集
• 调度定期任务
• 分发计算密集型操作

快速开始

最简可运行示例：

# celery_app.py
from celery import Celery
from kombu import Exchange, Queue
import os

app = Celery('myapp')

# Configuration
app.conf.update(
    broker_url=os.getenv('REDIS_URL', 'redis://localhost:6379/0'),
    result_backend=os.getenv('REDIS_URL', 'redis://localhost:6379/0'),
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='UTC',
    enable_utc=True,
    task_track_started=True,
    task_time_limit=30 * 60,  # 30 minutes
    task_soft_time_limit=25 * 60,  # 25 minutes
    broker_connection_retry_on_startup=True,
)

# Queue configuration
default_exchange = Exchange('tasks', type='direct')
app.conf.task_queues = (
// ... (see reference guides for full implementation)

参考指南

references/ 目录中的详细实现：

指南	内容
使用 Celery 和 Redis 的 Python	使用 Celery 和 Redis 的 Python
使用 Bull Queue 的 Node.js	使用 Bull Queue 的 Node.js
使用 Sidekiq 的 Ruby	使用 Sidekiq 的 Ruby
作业重试和错误处理	作业重试和错误处理
监控和可观测性	监控和可观测性

最佳实践

✅ 应该

• 使用任务超时，防止作业挂起
• 使用指数退避实现重试逻辑
• 确保任务具备幂等性
• 为关键任务设置作业优先级
• 监控队列深度和作业失败情况
• 记录作业执行详情
• 清理已完成的作业
• 设置合适的批次大小，以提高内存效率
• 对失败的作业使用死信队列
• 独立测试作业

❌ 不应该

• 在异步任务中使用同步操作
• 忽略作业失败
• 让任务依赖外部状态
• 使用无上限的重试
• 在作业数据中存储大型对象
• 忘记处理超时
• 在缺乏监控的情况下运行作业
• 在队列中使用阻塞操作
• 忘记跟踪作业进度
• 在一个作业中混合不相关的操作