celery-redis异步任务具体应用
·
Celery + Redis 异步任务队列实战指南
本文以商城项目为例，深入讲解 Celery + Redis 异步任务队列的架构设计与实现细节，并分析为何选择 Redis 而非 RabbitMQ 作为消息代理。
一、什么是异步任务队列？
在 Web 应用开发中，某些操作耗时较长，如果在请求中同步执行，会导致用户等待时间过长，严重影响用户体验。异步任务队列的核心思想是：将耗时任务从主请求中剥离，交给后台异步处理。
常见异步任务场景
| 场景 | 耗时原因 | 异步处理优势 |
|---|---|---|
| 发送短信/邮件 | 第三方 API 调用延迟 | 用户无需等待发送完成 |
| 图片/视频处理 | CPU 密集型计算 | 不阻塞 Web 服务器 |
| 静态页面生成 | 文件 I/O 操作 | 提高页面访问速度 |
| 数据统计分析 | 复杂 SQL 查询 | 避免数据库超时 |
| 第三方 API 调用 | 网络延迟不可控 | 提高系统稳定性 |
二、Celery 架构解析
2.1 核心组件
Celery异步任务队列架构
2.2 组件职责
| 组件 | 职责 | 本项目实现 |
|---|---|---|
| Producer | 创建并发送任务消息 | Django 视图/序列化器 |
| Broker | 存储任务消息队列 | Redis (db=2) |
| Worker | 从队列取出任务并执行 | Celery Worker 进程 |
| Result Backend | 存储任务执行结果 | Redis (db=3) |
三、项目实战：美多商城的 Celery 实现
3.1 项目目录结构
meiduo_mall/
├── celery_tasks/ # Celery 任务模块
│ ├── __init__.py
│ ├── main.py # Celery 应用入口
│ ├── config.py # Celery 配置文件
│ ├── sms/ # 短信任务
│ │ ├── __init__.py
│ │ ├── tasks.py
│ │ └── constants.py
│ ├── email/ # 邮件任务
│ │ ├── __init__.py
│ │ └── tasks.py
│ └── html/ # 静态页面生成任务
│ ├── __init__.py
│ └── tasks.py
└── meiduo_mall/
└── settings/
└── dev.py # Django 配置
3.2 Celery 应用初始化
文件：celery_tasks/main.py
from celery import Celery
import os
# 禁用 eventlet DNS 补丁，避免兼容性问题
os.environ['EVENTLET_NO_GREENDNS'] = 'yes'
# 设置 Django 配置模块，使 Celery 能读取 Django 配置
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'meiduo_mall.settings.dev')
# 创建 Celery 应用实例
celery_app = Celery('meiduo')
# 加载配置文件
celery_app.config_from_object('celery_tasks.config')
# 自动发现并注册任务模块
celery_app.autodiscover_tasks(['celery_tasks.sms'])
celery_app.autodiscover_tasks(['celery_tasks.email'])
celery_app.autodiscover_tasks(['celery_tasks.html'])
关键点解析：
DJANGO_SETTINGS_MODULE：让 Celery 能够访问 Django 的配置和模型Celery('meiduo')：创建名为 ‘meiduo’ 的 Celery 应用autodiscover_tasks：自动扫描指定模块中的任务函数
3.3 Celery 配置文件
文件：celery_tasks/config.py
# 消息代理地址 - Redis 数据库 2
broker_url = 'redis://192.168.26.149:6379/2'
# 结果存储地址 - Redis 数据库 3
result_backend = 'redis://192.168.26.149:6379/3'
# Worker 并发数量
worker_concurrency = 4
# 时区设置（可选）
timezone = 'Asia/Shanghai'
# 任务序列化格式
task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
Redis 数据库规划：
| 数据库编号 | 用途 | 说明 |
|---|---|---|
| db=2 | Celery Broker | 任务消息队列 |
| db=3 | Celery Result | 任务执行结果 |
| db=4 | 验证码 | 短信验证码 |
3.4 定义异步任务
示例一：短信发送任务
文件：celery_tasks/sms/tasks.py
from celery_tasks.main import celery_app
from celery_tasks.sms.yuntongxun.rl_sms_sdk import RLSMS
from celery_tasks.sms import constants
# 使用 @celery_app.task 装饰器定义任务
@celery_app.task(name='send_sms_code')
def send_sms_code(mobile, sms_code):
"""
发送短信验证码（异步任务）
Args:
mobile: 手机号
sms_code: 验证码
"""
RLSMS().send_template_sms(
mobile,
(sms_code, constants.SMS_CODE_REDIS_EXPIRES // 60),
1 #沙箱模式
)
示例二：邮件发送任务
文件：celery_tasks/email/tasks.py
from celery_tasks.main import celery_app
from django.core.mail import send_mail
from django.conf import settings
@celery_app.task(name='send_email')
def send_verify_email(to_email, verify_url):
"""
发送邮箱验证邮件（异步任务）
Args:
to_email: 收件人邮箱
verify_url: 验证链接
"""
subject = "美多商城邮箱验证"
html_message = f'''
<p>尊敬的用户您好！</p>
<p>感谢您使用美多商城。</p>
<p>您的邮箱为：{to_email}。请点击此链接激活您的邮箱：</p>
<p><a href="{verify_url}">{verify_url}</a></p>
'''
send_mail(
subject=subject,
message='',
from_email='美多商城<wc3445979735@163.com>',
recipient_list=[to_email],
html_message=html_message
)
示例三：静态页面生成任务
文件：celery_tasks/html/tasks.py
from celery_tasks.main import celery_app
from django.template import loader
from django.conf import settings
import os
from goods.utils import get_categories
from goods.models import SKU
@celery_app.task(name='generate_static_list_search_html')
def generate_static_list_search_html():
"""
生成静态的商品列表页和搜索结果页（异步任务）
"""
categories = get_categories()
context = {'categories': categories}
template = loader.get_template('list.html')
html_text = template.render(context)
file_path = os.path.join(
settings.GENERATED_STATIC_HTML_FILES_DIR,
'list.html'
)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(html_text)
@celery_app.task(name='generate_static_sku_detail_html')
def generate_static_sku_detail_html(sku_id):
"""
生成静态商品详情页面（异步任务）
Args:
sku_id: 商品 SKU ID
"""
categories = get_categories()
sku = SKU.objects.get(id=sku_id)
# ... 构建上下文数据
template = loader.get_template('detail.html')
html_text = template.render(context)
file_path = os.path.join(
settings.GENERATED_STATIC_HTML_FILES_DIR,
f'goods/{sku_id}.html'
)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(html_text)
3.5 触发异步任务
在视图中触发任务
# verifications/views.py
from celery_tasks.sms.tasks import send_sms_code
class SMSCodeView(APIView):
def get(self, request, mobile):
# 生成验证码
sms_code = '%06d' % random.randint(0, 999999)
# 存储到 Redis
redis_conn = get_redis_connection('verify_codes')
redis_conn.setex(f'sms_{mobile}', 300, sms_code)
# 异步发送短信 - 使用 .delay() 触发任务
send_sms_code.delay(mobile, sms_code)
return Response({'message': 'OK'})
在序列化器中触发任务
# users/serializers.py
from celery_tasks.email.tasks import send_verify_email
class EmailSerializer(serializers.ModelSerializer):
def update(self, instance, validated_data):
email = validated_data['email']
instance.email = email
instance.save()
# 生成验证链接
verify_url = instance.generate_email_verify_url()
# 异步发送邮件
send_verify_email.delay(email, verify_url)
return instance
在 Admin 中触发任务
# goods/admin.py
from celery_tasks.html.tasks import (
generate_static_list_search_html,
generate_static_sku_detail_html
)
class SKUAdmin(admin.ModelAdmin):
def save_model(self, request, obj, form, change):
obj.save()
# 商品保存后，异步生成静态页面
generate_static_sku_detail_html.delay(obj.id)
def delete_model(self, request, obj):
generate_static_sku_detail_html.delay(obj.id)
obj.delete()
3.6 启动 Celery Worker
# 进入项目目录
cd /path/to/meiduo_mall
# 启动 Worker（开发环境）
celery -A celery_tasks.main worker -l info
# 启动 Worker（生产环境，指定并发数）
celery -A celery_tasks.main worker -l info --concurrency=4
# 后台运行（Linux）
celery -A celery_tasks.main worker -l info --daemon
# 指定队列名称
celery -A celery_tasks.main worker -l info -Q default,celery
启动参数说明：
| 参数 | 说明 |
|---|---|
-A |
指定 Celery 应用模块 |
-l |
日志级别：debug/info/warning/error |
--concurrency |
Worker 并发进程数 |
-Q |
指定监听的队列 |
-n |
指定 Worker 名称 |
--daemon |
后台运行（Linux） |
四、为何选择 Redis 而非 RabbitMQ？
4.1 Broker 选型对比
| 特性 | Redis | RabbitMQ |
|---|---|---|
| 性能 | 内存操作，极高吞吐量 | 磁盘持久化，中等吞吐量 |
| 部署复杂度 | 简单，单进程 | 复杂，需要 Erlang 环境 |
| 内存占用 | 较低 | 较高 |
| 消息持久化 | 可选（默认内存） | 默认持久化 |
| 消息确认 | 简单 | 完善的 ACK 机制 |
| 优先级队列 | 不支持 | 支持 |
| 管理界面 | 需要额外工具 | 内置 Web 管理界面 |
| 学习曲线 | 平缓 | 陡峭 |
4.2 项目选择 Redis 的原因
原因一：已有 Redis 基础设施
项目中 Redis 已经被广泛使用：
# settings/dev.py
CACHES = {
"default": {...}, # 缓存
"session": {...}, # Session 存储
"verify_codes": {...}, # 验证码
"history": {...}, # 浏览记录
"cart": {...}, # 购物车
}
复用现有 Redis 服务，无需额外部署 RabbitMQ，降低运维成本。
原因二：性能足够满足需求
美多商城的任务特点：
| 任务类型 | 频率 | 耗时 | 可靠性要求 |
|---|---|---|---|
| 短信发送 | 中等 | 1-3秒 | 中 |
| 邮件发送 | 低 | 2-5秒 | 中 |
| 静态页面生成 | 低 | 1-10秒 | 高 |
Redis 的性能完全满足这些轻量级异步任务的需求。
原因三：开发环境友好
- Redis：Windows/Linux/Mac 都易于安装
- RabbitMQ：Windows 安装需要 Erlang 环境，配置复杂
4.3 何时应该选择 RabbitMQ？
| 场景 | 推荐 Broker |
|---|---|
| 高可靠性金融交易系统 | RabbitMQ |
| 消息不能丢失的关键业务 | RabbitMQ |
| 需要复杂路由规则 | RabbitMQ |
| 需要消息优先级 | RabbitMQ |
| 中小型 Web 应用 | Redis |
| 已有 Redis 基础设施 | Redis |
| 追求高性能吞吐量 | Redis |
| 开发测试环境 | Redis |
五、最佳实践总结
5.1 任务设计原则
# ✅ 好的设计：任务函数职责单一
@celery_app.task(name='send_sms_code')
def send_sms_code(mobile, sms_code):
RLSMS().send_template_sms(mobile, (sms_code, 5), 1)
# ❌ 不好的设计：任务函数包含过多业务逻辑
@celery_app.task(name='process_order')
def process_order(order_id):
order = Order.objects.get(id=order_id)
# 检查库存
# 扣减库存
# 发送短信
# 发送邮件
# 更新状态
# ... 太多职责
5.2 任务重试机制
@celery_app.task(
name='send_sms_code',
bind=True,
max_retries=3,
default_retry_delay=60
)
def send_sms_code(self, mobile, sms_code):
try:
RLSMS().send_template_sms(mobile, (sms_code, 5), 1)
except Exception as exc:
# 自动重试
raise self.retry(exc=exc)
5.3 任务结果查询
# 触发任务并获取任务 ID
result = send_sms_code.delay(mobile, sms_code)
task_id = result.id
# 查询任务状态
from celery.result import AsyncResult
task_result = AsyncResult(task_id)
print(task_result.status) # PENDING/STARTED/SUCCESS/FAILURE
print(task_result.result) # 任务返回值
print(task_result.traceback) # 错误堆栈（如果失败）
5.4 定时任务配置
# celery_tasks/config.py
from celery.schedules import crontab
beat_schedule = {
# 每5分钟执行一次
'generate-static-index-every-5-min': {
'task': 'generate_static_list_search_html',
'schedule': 300.0, # 秒
},
# 每天凌晨1点执行
'cleanup-every-day': {
'task': 'cleanup_expired_data',
'schedule': crontab(hour=1, minute=0),
},
}
启动 Beat 调度器：
celery -A celery_tasks.main beat -l info
六、常见问题与解决方案
Q1：任务执行失败如何排查？
# 查看 Worker 日志
celery -A celery_tasks.main worker -l debug
# 检查 Redis 中的任务队列
redis-cli -n 2
> LRANGE celery 0 -1
Q2：如何保证任务不丢失？
# config.py
task_acks_late = True # 任务执行成功后才确认
task_reject_on_worker_lost = True
task_default_delivery_mode = 'persistent' # 持久化
Q3：如何监控任务执行情况？
推荐使用 Flower 监控工具：
pip install flower
celery -A celery_tasks.main flower
# 访问 http://localhost:5555
Q4：Windows 下 Celery 4.x+ 无法运行？
# 安装 eventlet
pip install eventlet
# 使用 eventlet 启动
celery -A celery_tasks.main worker -l info -P eventlet
七、总结
Celery + Redis 异步任务队列方案在美多商城项目中的成功实践证明：
- 架构简洁：复用现有 Redis 服务，无需额外中间件
- 性能优异：内存级消息传递，高吞吐量
- 开发友好：配置简单，学习成本低
- 运维便捷：单服务管理，监控方便
对于中小型 Web 应用，特别是已有 Redis 基础设施的项目，Celery + Redis 是一个高效、实用的异步任务解决方案。
参考资料
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
31
0- 0
已为社区贡献5条内容
所有评论(0)