Celery异步的分布式任务调度理解

时间：2021-08-23 09:05:58

什么是Celery呢？

Celery是一个用Python开发的异步的分布式任务调度模块。

Celery本身不包含消息服务，使用第三方消息服务，也就是Broker，来传递任务，目前支持的有Rebbimq，Redis，数据库以及其他的一些比如Amazon SQS,Monogdb和IronMQ 。

Celery支持同步和异步执行两种模式。同步模式为任务调用方等待任务执行完成，这种方式等同于RPC(Remote Procedure Call)， 异步方式为任务在后台执行，调用方调用后就去做其他工作，之后再根据需要来查看任务结果。Celery自己没有实现消息队列，而是直接已存在的消息队列作为Broker角色。官方推荐的Broker为 RabbitMQ ，除此之外，Redis、Beanstalkd、MongoDB等也都支持，具体可参考 官方文档。

Celery整体架构可以理解为下图:

Celery异步的分布式任务调度理解

整体上包括三个角色:

  • Celery client: 这是任务生产者，它负责将任务发送到Broker中。
  • Broker: Broker负责将任务分发给相应的celery worker。
  • Celery worker: 这是任务的执行者，完成相应的业务逻辑，在具体实现上体现为Python函数。

下面说一下Celery实例:

使用Celery完成发送短信:

这里使用redis作为消息队列（任务队列）

　　关于Redis和Rebbimq 消息队列的比较：https://www.cnblogs.com/lzc978/articles/10291597.html

整个celery服务器实时监听任务队列逻辑

Celery异步的分布式任务调度理解

0. 安装环境

0.1 pip安装Celery框架

pip install celery

0.2 启动Celery

celery -A 应用的包路路径 worker -l info  # ‘-l: left’, 'info: 日志等级'

1. 代码结构

Celery异步的分布式任务调度理解

1.1 config.py配置文件

# celery配置文件

# 指定任务队列列的位置 使用redis做为broker任务队列

broker_url = "redis://192.168.103.210/7"

1.2 main.py celery分布式异步任务队列启动文件

# celery启动文件

from celery import Celery

# 创建celery实例例

celery_app = Celery('file')

# 加载celery配置

celery_app.config_from_object('celery_tasks.config')

# 自动注册celery任务

celery_app.autodiscover_tasks(['celery_tasks.sms'])

2. Celery异步任务的定义

2.1 sms.tasks.py  异步任务/将被放在redis的任务队列(broker)中等待触发执行

# 发送短信的异步任务

from .yuntongxun.sms import CCP

from . import constants

from celery_tasks.main import celery_app

# 装饰器器将send_sms_code装饰为异步任务,并设置别名

@celery_app.task(name='send_sms_code')

def send_sms_code(mobile, sms_code):

　　"""

　　发送短信异步任务

　　:param mobile: 手机号

　　:param sms_code: 短信验证码

　　:return: None

　　"""

　　CCP().send_template_sms(mobile, [sms_code, constants.SMS_CODE_REDIS_EXPIRES // 60], constants.SEND_SMS_TEMPLATE_ID)

3. 开启Celery

celery -A celery_tasks.main worker -l info

4. 执行Celery异步任务（触发worker）

4.1 views.py  视图逻辑，即触发执行worker(任务队列中sms)的业务逻辑

# 生成和发送短信验证码

sms_code = '%06d' % random.randint(0,999999)

# CCP().send_template_sms(mobile,[sms_code, constants.SMS_CODE_REDIS_EXPIRES // 60], 1)

# celery异步发送短信

send_sms_code.delay(mobile,sms_code)

至此，一个简单的异步分布式Celery(异步任务)服务器搭建完成

相关文章