Python开发异步任务Celery的使用教程！

1. 生产者消费者设计模式

最常用的解耦方式之一，寻找中间人(broker)搭桥，保证两个业务没有直接关联。
我们称这一解耦方式为：生产者消费者设计模式

2.中间人broker

示例：此处演示Redis数据库作为中间人broker
Celery需要一种解决消息的发送和接受的方式，我们把这种用来存储消息的的中间装置叫做message broker, 也可叫做消息中间人。
作为中间人，我们有几种方案可选择：

1.RabbitMQ

RabbitMQ是一个功能完备，稳定的并且易于安装的broker. 它是生产环境中最优的选择。
使用RabbitMQ的细节参照以下链接：http://docs.celeryproject.org/en/latest/getting-started/brokers/rabbitmq.html#broker-rabbitmq
如果使用的是Ubuntu或者Debian发行版的Linux，可以直接通过命令安装RabbitMQ:

sudo apt-get install rabbitmq-server

安装完毕之后，RabbitMQ-server服务器就已经在后台运行。
如果用的并不是Ubuntu或Debian, 可以在以下网址：
http://www.rabbitmq.com/download.html
去查找自己所需要的版本软件。

2.Redis

Redis也是一款功能完备的broker可选项，但是其更可能因意外中断或者电源故障导致数据丢失的情况。
关于是由那个Redis作为Broker，可访下面网址：http://docs.celeryproject.org/en/latest/getting-started/brokers/redis.html#broker-redis

1. Celery介绍

Celery介绍：
一个简单、灵活且可靠、处理大量消息的分布式系统，可以在一台或者多台机器上运行。
单个 Celery 进程每分钟可处理数以百万计的任务。
通过消息进行通信，使用消息队列（broker）在客户端和消费者之间进行协调。
安装Celery：

$ pip install -U Celery

Celery官方文档
2. 创建Celery实例并加载配置

1.定义Celery包

2.创建Celery实例

celery_tasks.main.py

# celery启动文件
from celery import Celery

# 为celery使用django配置文件进行设置
import os
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "xxx.settings")

# 创建celery实例
celery_app = Celery('celery_tasks')

3.加载Celery配置

celery_tasks.config.py

broker_url = "redis://127.0.0.1/14"
result_backend = "redis://127.0.0.1/15"

celery_tasks.main.py

# celery启动文件
from celery import Celery


# 创建celery实例
celery_app = Celery('xxx')
# 加载celery配置
celery_app.config_from_object('celery_tasks.config')

3. 定义发送短信任务

1.注册任务：celery_tasks.main.py

# celery启动文件
from celery import Celery


# 创建celery实例
celery_app = Celery('celery_tasks')
# 加载celery配置
celery_app.config_from_object('celery_tasks.config')
# 自动注册celery任务
celery_app.autodiscover_tasks(['celery_tasks.sms'])

2.定义任务：celery_tasks.sms.tasks.py
tasks

from apps.verifications import constants
from celery_tasks.main import celery_app
from libs.yuntongxun.sms import CCP
import logging
logger = logging.getLogger('django')

# bind：保证task对象会作为第一个参数自动传入
# name：异步任务别名
# retry_backoff：异常自动重试的时间间隔 第n次(retry_backoff×2^(n-1))s
# max_retries：异常自动重试次数的上限
@celery_app.task(bind=True, name='send_sms_code', retry_backoff=3)
def send_sms_code(self, mobile, sms_code):
"""
发送短信异步任务
:param mobile: 手机号
:param sms_code: 短信验证码
:return: 成功0 或 失败-1
"""
try:
send_ret = CCP().send_template_sms(mobile, [sms_code, constants.SMS_CODE_REDIS_EXPIRES // 60], constants.SEND_SMS_TEMPLATE_ID)
except Exception as e:
logger.error(e)
# 有异常自动重试三次
raise self.retry(exc=e, max_retries=3)
if send_ret != 0:
# 有异常自动重试三次
raise self.retry(exc=Exception('发送短信失败'), max_retries=3)

return send_ret

4. 启动Celery服务

$ cd ~/xxx_project/xxx
$ celery -A celery_tasks.main worker -l info

-A指对应的应用程序, 其参数是项目中 Celery实例的位置。
worker指这里要启动的worker。
-l指日志等级，比如info等级。

 

5. 调用发送短信任务

# 发送短信验证码
# CCP().send_template_sms(mobile,[sms_code, constants.SMS_CODE_REDIS_EXPIRES // 60], constants.SEND_SMS_TEMPLATE_ID)
# Celery异步发送短信验证码

send_sms_code.delay(mobile, sms_code)

 

6. 补充celery worker的工作模式
默认是进程池方式，进程数以当前机器的CPU核数为参考，每个CPU开四个进程。
如何自己指定进程数：

celery worker -A proj --concurrency=4
如何改变进程池方式为协程方式：

celery worker -A proj --concurrency=1000 -P eventlet -c 1000

# 安装eventlet模块
$ pip install eventlet

# 启用 Eventlet 池
$ celery -A celery_tasks.main worker -l info -P eventlet -c 1000