Django项目如何用Celery实现异步任务完整步骤

需求

django中使用celery异步发送信息或者邮箱

本文介绍django中如何使用celery，与常规的配置方法有所不同，本文是使用集中管理的方式配置celery，这样便于维护与管理，又不与django配置相混淆。

假设你的机器上已经安装好了redis数据库。

一、安装

    pip install django-celery
    pip install django-redis
    pip install celery
    pip install redis

安装好之后在INSTALLED_APPS中增加djcelery应用：

注意版本之间的兼容性，之前就踩过很多坑，运行时发现报错，就是因为版本之间不兼容导致，本例中使用的版本：

其他测试可用的版本：

Django==3.2
django-celery== 3.1.17
django-redis== 4.11.0
celery== 4.4.2
redis==3.5.3

二、文件结构

我的项目名称叫hrms，首先在项目根目录下新建celery_tasks包，作为celery服务的根目录。

• config.py是celery的配置文件；
• main.py是celery服务启动的入口文件；
• mail包是发邮件的任务目录
• sms包是发信息的任务目录

三、config.py文件

目前只做了两个简单配置，更复杂的配置比如日志、定时任务等会在以后介绍。

    指定任务队列的位置
    BROKER_URL = "redis://127.0.0.1/0"
    指定消息执行结果的位置
    CELERY_RESULT_BACKEND =  "redis://127.0.0.1/0"

四、main.py文件

    from celery import Celery
    import os
    
    # 为celery使用django配置文件进行设置，注意main.py和settings.py的相对位置关系
    if not os.getenv('DJANGO_SETTINGS_MODULE'):
        os.environ['DJANGO_SETTINGS_MODULE'] = 'hrms.settings.settings'   
        
    # 创建celery对象,并起别名
    celery_app = Celery('hrms')  

	#从配置文件加载配置
    celery_app.config_from_object('celery_tasks.config')  
    
    #注册celery任务
     celery_app.autodiscover_tasks(['celery_tasks.mail', 'celery_tasks.sms'])

五、tasks.py文件

from django.conf import settings
from django.core.mail import send_mail
from celery_tasks.main import celery_app


@celery_app.task(name='send_verify_email')
def send_email(recipient_list,verify_url):
    # send_mail(subject, message, from_email, recipient_list, html_message=None)
    subject='xxxx有限验证'
    from_email=settings.EMAIL_FROM
    html_message='
尊敬的用户您好！
' 
                   '
感谢您使用xxxx。
' 
                   '
您的邮箱为：%s 。请点击此链接激活您的邮箱：
' 
                   '
%s
' % (recipient_list, verify_url, verify_url)
                   
    send_mail(subject, "", from_email, [recipient_list], html_message=html_message)

from celery_tasks.email.tasks import send_email

    def update(self, instance, validated_data):
        # 调用models中的方法生成url
        verify_url = instance.generate_verify_email_url()

        # celery发送邮件
        send_email.delay(instance.email, verify_url)

        return instance

python manage.py runserver  启动django
celery worker -A celery_tasks.main -l debug  启动celery worker 服务