celery-redis异步任务具体应用

Celery + Redis 异步任务队列实战指南

本文以商城项目为例，深入讲解 Celery + Redis 异步任务队列的架构设计与实现细节，并分析为何选择 Redis 而非 RabbitMQ 作为消息代理。

一、什么是异步任务队列？

在 Web 应用开发中，某些操作耗时较长，如果在请求中同步执行，会导致用户等待时间过长，严重影响用户体验。异步任务队列的核心思想是：将耗时任务从主请求中剥离，交给后台异步处理。

常见异步任务场景

二、Celery 架构解析

2.1 核心组件

Celery异步任务队列架构

2.2 组件职责

三、项目实战：美多商城的 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

fromceleryimportCeleryimportos# 禁用 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'])

关键点解析：

1. DJANGO_SETTINGS_MODULE：让 Celery 能够访问 Django 的配置和模型
2. Celery('meiduo')：创建名为 ‘meiduo’ 的 Celery 应用
3. autodiscover_tasks：自动扫描指定模块中的任务函数

3.3 Celery 配置文件

文件：celery_tasks/config.py

# 消息代理地址 - Redis 数据库 2broker_url='redis://192.168.26.149:6379/2'# 结果存储地址 - Redis 数据库 3result_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 数据库规划：

3.4 定义异步任务

示例一：短信发送任务

文件：celery_tasks/sms/tasks.py

fromcelery_tasks.mainimportcelery_appfromcelery_tasks.sms.yuntongxun.rl_sms_sdkimportRLSMSfromcelery_tasks.smsimportconstants# 使用 @celery_app.task 装饰器定义任务@celery_app.task(name='send_sms_code')defsend_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

fromcelery_tasks.mainimportcelery_appfromdjango.core.mailimportsend_mailfromdjango.confimportsettings@celery_app.task(name='send_email')defsend_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

fromcelery_tasks.mainimportcelery_appfromdjango.templateimportloaderfromdjango.confimportsettingsimportosfromgoods.utilsimportget_categoriesfromgoods.modelsimportSKU@celery_app.task(name='generate_static_list_search_html')defgenerate_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')withopen(file_path,'w',encoding='utf-8')asf:f.write(html_text)@celery_app.task(name='generate_static_sku_detail_html')defgenerate_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')withopen(file_path,'w',encoding='utf-8')asf:f.write(html_text)

3.5 触发异步任务

在视图中触发任务

# verifications/views.pyfromcelery_tasks.sms.tasksimportsend_sms_codeclassSMSCodeView(APIView):defget(self,request,mobile):# 生成验证码sms_code='%06d'%random.randint(0,999999)# 存储到 Redisredis_conn=get_redis_connection('verify_codes')redis_conn.setex(f'sms_{mobile}',300,sms_code)# 异步发送短信 - 使用 .delay() 触发任务send_sms_code.delay(mobile,sms_code)returnResponse({'message':'OK'})

在序列化器中触发任务

# users/serializers.pyfromcelery_tasks.email.tasksimportsend_verify_emailclassEmailSerializer(serializers.ModelSerializer):defupdate(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)returninstance

在 Admin 中触发任务

# goods/admin.pyfromcelery_tasks.html.tasksimport(generate_static_list_search_html,generate_static_sku_detail_html)classSKUAdmin(admin.ModelAdmin):defsave_model(self,request,obj,form,change):obj.save()# 商品保存后，异步生成静态页面generate_static_sku_detail_html.delay(obj.id)defdelete_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-Acelery_tasks.main worker-linfo# 启动 Worker（生产环境，指定并发数）celery-Acelery_tasks.main worker-linfo--concurrency=4# 后台运行（Linux）celery-Acelery_tasks.main worker-linfo--daemon# 指定队列名称celery-Acelery_tasks.main worker-linfo-Qdefault,celery

启动参数说明：

四、为何选择 Redis 而非 RabbitMQ？

4.1 Broker 选型对比

4.2 项目选择 Redis 的原因

原因一：已有 Redis 基础设施

项目中 Redis 已经被广泛使用：

# settings/dev.pyCACHES={"default":{...},# 缓存"session":{...},# Session 存储"verify_codes":{...},# 验证码"history":{...},# 浏览记录"cart":{...},# 购物车}

复用现有 Redis 服务，无需额外部署 RabbitMQ，降低运维成本。

原因二：性能足够满足需求

美多商城的任务特点：

Redis 的性能完全满足这些轻量级异步任务的需求。

原因三：开发环境友好

• Redis：Windows/Linux/Mac 都易于安装
• RabbitMQ：Windows 安装需要 Erlang 环境，配置复杂

4.3 何时应该选择 RabbitMQ？

五、最佳实践总结

5.1 任务设计原则

# ✅ 好的设计：任务函数职责单一@celery_app.task(name='send_sms_code')defsend_sms_code(mobile,sms_code):RLSMS().send_template_sms(mobile,(sms_code,5),1)# ❌ 不好的设计：任务函数包含过多业务逻辑@celery_app.task(name='process_order')defprocess_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)defsend_sms_code(self,mobile,sms_code):try:RLSMS().send_template_sms(mobile,(sms_code,5),1)exceptExceptionasexc:# 自动重试raiseself.retry(exc=exc)

5.3 任务结果查询

# 触发任务并获取任务 IDresult=send_sms_code.delay(mobile,sms_code)task_id=result.id# 查询任务状态fromcelery.resultimportAsyncResult task_result=AsyncResult(task_id)print(task_result.status)# PENDING/STARTED/SUCCESS/FAILUREprint(task_result.result)# 任务返回值print(task_result.traceback)# 错误堆栈（如果失败）

5.4 定时任务配置

# celery_tasks/config.pyfromcelery.schedulesimportcrontab 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-Acelery_tasks.main beat-linfo

六、常见问题与解决方案

Q1：任务执行失败如何排查？

# 查看 Worker 日志celery-Acelery_tasks.main worker-ldebug# 检查 Redis 中的任务队列redis-cli-n2>LRANGE celery0-1

Q2：如何保证任务不丢失？

# config.pytask_acks_late=True# 任务执行成功后才确认task_reject_on_worker_lost=Truetask_default_delivery_mode='persistent'# 持久化

Q3：如何监控任务执行情况？

推荐使用Flower监控工具：

pipinstallflower celery-Acelery_tasks.main flower# 访问 http://localhost:5555

Q4：Windows 下 Celery 4.x+ 无法运行？

# 安装 eventletpipinstalleventlet# 使用 eventlet 启动celery-Acelery_tasks.main worker-linfo-Peventlet

七、总结

Celery + Redis 异步任务队列方案在美多商城项目中的成功实践证明：

1. 架构简洁：复用现有 Redis 服务，无需额外中间件
2. 性能优异：内存级消息传递，高吞吐量
3. 开发友好：配置简单，学习成本低
4. 运维便捷：单服务管理，监控方便

对于中小型 Web 应用，特别是已有 Redis 基础设施的项目，Celery + Redis 是一个高效、实用的异步任务解决方案。

参考资料

• Celery 官方文档
• Redis 官方文档
• Django Celery 最佳实践