1.Celery简单的使用

pip install celery==4.4.7
pip install eventlet==0.26.1

celery，分布式异步任务队列
eventlet，并发网络库 gevent 协程库

1. 创建一个laufing.py文件里面写入代码

#实例化对象
from celery import Celery

# 第一个参数worker name 
# broker 代理，消息中间件
# "myworker", broker="redis://:密码(没有密码可以不用写)@localhost:6379/选择库"
app = Celery("myworker", broker="redis://:[email protected]:6379/4")
#也可以app.conf.broker_url = "xxxx"

#创建任务函数
@app.task
def task1():
    print("正在执行任务...")


#window
# 端口输入命令
celery -A laufing worker -l info -P eventlet   

ipython中模拟发送任务：在laufing.py的同级目录打开一个cmd命令行。

2. Celery存储任务结果
创建一个laufing.py文件，并创建worker

from celery  import Celery

app = Celery("worker2", broker="redis://:[email protected]:6379/4", backend="redis://:[email protected]:6379/5")

@app.task
def task1(a, b):
	print("简单求和")
	return a + b

@app.task
def task2(a, b):
	print("2-s后求和")
	time.sleep(20)
	return a + b

前台启动worker

#window
celery -A laufing worker -l info -P eventlet

发送任务，并获取结果
ipython中测试如下：

from laufing import task1, task2

r1 = task1.delay(3, 5)  # 发送任务
r1.result

r2 = task2.delay(4, 7) #发送任务，并立即执行下一行，不会阻塞
r2.result   # worker来执行task，返回结果后才可以拿到

2. celery异步发送短信

1. 在django项目中的主应用里，创建config.py文件里面写入代码

broker_url = 'redis://:@127.0.0.1:6379/2'
result_backend = 'redis://:@127.0.0.1:6379/3'

2. 主应用下在创建一个celery.py文件写入代码

from celery import Celery
from django.conf import settings
import os


# 设置环境变量
os.environ.setdefault("DJANGO_SETTINGS_MODULE", 'num5.settings')

# 实例化celery对象
app = Celery('cc')
# 通过配置文件，进行配置
app.config_from_object('num5.config')

# 让app自动发现任务函数
app.autodiscover_tasks(settings.INSTALLED_APPS)

3. 在子应用中创建一个tasks.py文件封装函数

from num5.celery import app
from ronglian_sms_sdk import SmsSDK
import json

accId = '8aaf07088185853e01818a78d71c0195'  # 主账户ID
accToken = '982f2e6dd6a54e6f83d014bc22ac7206'  # 账户授权令牌
appId = '8aaf07088185853e01818a78d80e019c'  # 默认


# Create your tests here.

# 定义发送短信的函数
@app.task
def sms_message(sms_code, mobile, expire=5):
    """ :param sms_code: 要发送的验证码 :param mobile: 发送的手机号 :param expire: 过期时间 :return: """
    # 实例化sdk对象
    sdk = SmsSDK(accId=accId, accToken=accToken, appId=appId)

    # 准备数据
    tid = '1'
    datas = ("%s" % sms_code, "%s" % expire)

    # 发送短信
    res = sdk.sendMessage(tid=tid, mobile=mobile, datas=datas)

    # 解析json字符串
    data = json.loads(res)
    print('容联云响应的数据:', data)

    # if data.get('statusCode') == '000000':
    # return True
    #
    # return False

    return data

5. 去views中写入类视图

from users.tasks import sms_message
import random, re

# 验证码
class SmsCodeAPIView(APIView):
    # 发送短信验证码
    def get(self, request):
        mobile = request.query_params.get('mobile')
        sms_code = random.randint(1000, 9999)

        # 判断手机号
        ser = re.findall(r'^1[3-9][0-9]{9}$', mobile)
        if not ser:
            return Response({
    'code': 400, 'msg': '请输入正确的手机号'})

        # 发送短信
        sms_message.delay(sms_code, mobile)
		
		# 这里不用等待发送的短信可以直接执行下一条命令
        r = redis.Redis(host='127.0.0.1', port=6379)
        r.set(mobile, sms_code, ex=(60 * 60 * 2))
        return Response({
    'code': 200, 'msg': '已发送请注意查收'})

6. 运行终端，在新建一个终端运行一下命令

celery -A num5 worker -l info -P eventlet