win10上django框架中使用celery4.4.2版本及redis实现任务调用

1.安装

pip install celery==4.4.2

2.在项目中创建一个自己的mycelery包用于存放任务及配置等内容

3.在config.py文件中指定任务队列及结果队列地址

# 任务队列的链接地址
broker_url = 'redis://127.0.0.1:6379/14'

# 结果队列的链接地址
result_backend = 'redis://127.0.0.1:6379/15'

4.在main.py中创建celery应用等

from celery import Celery
import os

# os.environ.setdefault('FORKED_BY_MULTIPROCESSING', '1')
# 1. 初始化Celery应用对象
app = Celery()

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'yd_api.settings.dev')
import django

django.setup()

# 2. 加载初始化配置
app.config_from_object("mycelery.config")

# 3. 注册异步任务
app.autodiscover_tasks(['mycelery.sms'])

# 4. 在终端启动celery，win10下需要采用协程来运行
pip install eventlet
celery -A mycelery.main worker -l info -P eventlet

5.在sms/tasks.py中指定任务方法

from mycelery.main import app
from .yuntongxun.sms import CCP
from yd_api.settings import constants
import logging

log = logging.getLogger('django')


@app.task(name="send_sms")
def send_sms(mobile, sms_code):
    ccp = CCP()
    ret = ccp.send_template_sms(mobile, [sms_code, constants.SMS_EXPIRE_TIME // 60], constants.SMS_TEMPLATE_ID)
    # print(f"ret={ret}")
    if not ret:
        log.error("用户注册短信发送失败！手机号：%s" % mobile)
    # print("mobile=%s, sms_code=%s", mobile, sms_code)

6.在views.py中调用

7.在终端执行