Django-Celery-Redis异步发短信

冯翔
2023-12-01

首先因为异步版本问题要统一,否则会运行不起来,不兼容!

python == 3.7.6

django == 2.1.8

celery == 4.4.7

redis == 3.5.3

eventlet == 0.26.1

什么是celery
Celery是一个功能完备即插即用的任务队列。它使得我们不需要考虑复杂的问题,使用非常简单。celery适用异步处理问题,当发送邮件、或者文件上传, 图像处理等等一些比较耗时的操作,我们可将其异步执行,这样用户不需要等待很久,提高用户体验。 celery的特点是:

简单,易于使用和维护,有丰富的文档。
高效,单个celery进程每分钟可以处理数百万个任务。
灵活,celery中几乎每个部分都可以自定义扩展。
celery核心
1、Task
任务(Task)就是你要做的事情,例如一个注册流程里面有很多任务,给用户发验证邮件就是一个任务,这种耗时任务可以交给Celery去处理,还有一种任务是定时任务,比如每天定时统计网站的注册人数,这个也可以交给Celery周期性的处理。

2、Broker
Broker 的中文意思是经纪人,指为市场上买卖双方提供中介服务的人。在Celery中它介于生产者和消费者之间经纪人,这个角色相当于数据结构中的队列。例如一个Web系统中,生产者是处理核心业务的Web程序,业务中可能会产生一些耗时的任务,比如短信,生产者会将任务发送给 Broker,就是把这个任务暂时放到队列中,等待消费者来处理。消费者是 Worker,是专门用于执行任务的后台服务。Worker 将实时监控队列中是否有新的任务,如果有就拿出来进行处理。Celery 本身不提供队列服务,一般用 Redis 或者 RabbitMQ 来扮演 Broker 的角色

3、Worker
Worker 就是那个一直在后台执行任务的人,也称为任务的消费者,它会实时地监控队列中有没有任务,如果有就立即取出来执行。

4、Beat
Beat 是一个定时任务调度器,它会根据配置定时将任务发送给 Broker,等待 Worker 来消费。

5、Backend
Backend 用于保存任务的执行结果,每个任务都有返回值,比如发送邮件的服务会告诉我们有没有发送成功,这个结果就是存在Backend中,当然我们并不总是要关心任务的执行结果。

Broker选择
Celery需要一种解决消息的发送和接受的方式,我们把这种用来存储消息的的中间装置叫做message broker, 也可叫做消息中间人。 作为中间人,我们有几种方案可选择:

RabbitMQ
RabbitMQ是一个功能完备,稳定的并且易于安装的broker. 它是生产环境中最优的选择。

使用RabbitMQ的细节参照以下链接: http://docs.celeryproject.org/en/latest/getting-started/brokers/rabbitmq.html#broker-rabbitmq

Redis
Redis也是一款功能完备的broker可选项,但是其更可能因意外中断或者电源故障导致数据丢失的情况。

Redis作为Broker,可访下面网址: http://docs.celeryproject.org/en/latest/getting-started/brokers/redis.html#broker-redis

将redis发布订阅模式用做消息队列和rabbitmq的区别:
可靠性
redis:没有相应的机制保证消息的可靠消费,如果发布者发布一条消息,而没有对应的订阅者的话,这条消息将丢失, 不会存在内存中;rabbbitmq: 具有消息消费确认机制,如果发布一条消息,还没有消费者消费该队列,那么这条消息将一直存放在队列中,直到有消费者消费了该条消息,以此可以保证消息的可靠消费。

实时性
redis实时性高,redis作为高效的缓存服务器,所有数据都存在在服务器中,所以它具有更高的实时性消费者负载均衡;rabbitmq队列可以被多个消费者同时监控消费,但是每一条消息只能被消费一次,由于rabbitmq的消费确认机制,因此它能够根据消费者的消费能力而调整它的负载,redis发布订阅模式,一个队列 可以被多个消费者同时订阅,当有消息到达时,会将该消息依次发送给每个订阅者;

持久性
redis:redis的持久化是针对 于整个redis缓存的内容,它有RDB和AOF两个持久化方式,可以将整个redis实例持久化到磁盘,以此来做数据备份,防止异常情况下导致数据丢失。rabbitmq:队列、消息都可以选择性持久化,持久化粒度更小,更灵活;队列监控rabbitmq实现了后台监控平台,可以在该平台上看到所有创建的队列的详细情况,良好的台后管理平台可以方便我们更好的使用;redis没有所谓的监控平台。

总结:

redis:轻量级、低延迟,高并发、低可靠性;

rabbitmq:重量级、高可靠,异步,不保证实时
在Django中使用celery
在django项目SylProject/下创建celery.py文件,配置以下内容:

# celery.py文件
import os
from celery import Celery

# 把celery和django进行组合,识别和加载django的配置文件
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "SylProject.settings")

# 创建celery实例
app = Celery('SylProject')

# 指定celery消息队列的配置
app.config_from_object('SylProject.config', namespace='CELERY')

# 从所有的django-app中加载任务
app.autodiscover_tasks()



@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

在django项目SylProject/下创建config.py文件,配置以下内容:

# 消息中间人设置
broker_url = 'redis://127.0.0.1:6379/15'
# 结果存储设置
result_backend = 'redis://127.0.0.1:6379/14'

在django项目SylProject/下__init__.py中写入以下内容:

# 绝对引用,使我们的celery模块不会与原始的celery冲突
from __future__ import absolute_import, unicode_literals
# 加入绝对引入以后,导入当前模块下的内容方法: from xx import xx as xx
from .celery import app as celery_app

__all__ = ('celery_app',)

在django的verificationsapp中创建tasks.py文件,在这里写入异步任务内容:

import json

from ronglian_sms_sdk import SmsSDK

accId = '8a216da875e463e00175eab36c08022f'
accToken = 'fdf64f8c7c474d2a8fbb3f09da2991b5'
appId = '8a216da875e463e00175eab36cfc0235'

from SylProject.celery import app

@app.task
def send_message(phone, number):
    sdk = SmsSDK(accId, accToken, appId)
    tid = '1'  # '1'  这是一个测试模板
    mobile = phone

    datas = (number, '5')
    resp = sdk.sendMessage(tid, mobile, datas)
    print(resp)
    resp_body = json.loads(resp)

django的verificationsapp/views中

from rest_framework.views import APIView
from rest_framework.response import Response
from verificationsapp.tasks import send_message
from django_redis import get_redis_connection

class SendMSCodeView(APIView):
    def post(self, request):
        # 获取数据
        phone = request.data.get('phone')
        image_code = request.data.get('image_code')
        image_code_uuid = request.data.get('image_code_uuid')
        # 校验数据
        if not all([phone, image_code, image_code_uuid]):
            return Response({'msg': '数据不完整', 'code': 400})
        # 验证图片验证码是否正确
        redis_client = get_redis_connection('img_code')
        redis_image_code = redis_client.get(image_code_uuid).decode()

        # 判断用户输入的验证码与redis数据库的验证码
        if image_code.lower() != redis_image_code.lower():
            return Response({'msg': '图片验证码不正确', 'code': 400})

        number = random.randint(100000, 999999)  # 生成随机6位数验证码

        # data = {'code': number}
        # send_sms(phone, data)
        # 删除redis里的image_code,保存phone_code

        send_message.delay(phone, number)
        # send_message(phone, number)  # 向容联云发送短信
        # pipeline管道:作用就是把多个命令放在一起来执行
        pl = redis_client.pipeline()
        pl.setex(phone, 60 * 5, number)  # 把验证码存入redis
        pl.delete(image_code_uuid)
        pl.execute()

        return Response({'code': 0, 'msg': '短信发送成功'})

启动celery

打开pycharm下的Terminal,进入到项目根目录下,使用eventlet启动celery
celery -A SylProject worker -l info -P eventlet
 类似资料: