欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

Python celery原理及运行流程解析

程序员文章站 2022-07-02 20:52:41
celery简介celery是一个基于分布式消息传输的异步任务队列,它专注于实时处理,同时也支持任务调度。它的执行单元为任务(task),利用多线程,如eventlet,等,它们能被并发地执行在单个或...

celery简介

celery是一个基于分布式消息传输的异步任务队列,它专注于实时处理,同时也支持任务调度。它的执行单元为任务(task),利用多线程,如eventlet,等,它们能被并发地执行在单个或多个职程服务器(worker servers)上。任务能异步执行(后台运行)或同步执行(等待任务完成)。

在生产系统中,celery能够一天处理上百万的任务。它的完整架构图如下:

Python celery原理及运行流程解析

组件介绍:

  • producer:调用了celery提供的api、函数或者装饰器而产生任务并交给任务队列处理的都是任务生产者。
  • celery beat:任务调度器,beat进程会读取配置文件的内容,周期性地将配置中到期需要执行的任务发送给任务队列。
  • broker:消息代理,又称消息中间件,接受任务生产者发送过来的任务消息,存进队列再按序分发给任务消费方(通常是消息队列或者数据库)。celery目前支持rabbitmq、redis、mongodb、beanstalk、sqlalchemy、zookeeper等作为消息代理,但适用于生产环境的只有rabbitmq和redis, 官方推荐 rabbitmq。
  • celery worker:执行任务的消费者,通常会在多台服务器运行多个消费者来提高执行效率。
  • result backend:任务处理完后保存状态信息和结果,以供查询。celery默认已支持redis、rabbitmq、mongodb、django orm、sqlalchemy等方式。

工作原理

它的基本工作就是管理分配任务到不同的服务器,并且取得结果。至于说服务器之间是如何进行通信的?这个celery本身不能解决。所以,rabbitmq作为一个消息队列管理工具被引入到和celery集成,负责处理服务器之间的通信任务。和rabbitmq的关系只是在于,celery没有消息存储功能,他需要介质,比如rabbitmq、redis、mysql、mongodb 都是可以的。推荐使用rabbitmq,他的速度和可用性都很高。

celery安装及使用

1、安装celery

pip install celery

2、查看完整可用命令选项

celery worker --help

3、创建一个工程项目project,然后再项目内创建一个celery_tasks异步任务列表。如图:

Python celery原理及运行流程解析

4、首先是celery_tasks异步任务主程序main.py,代码如下:

from celery import celery
# 生成celery应用
celery_app = celery("caicai")
# 加载配置文件
celery_app.config_from_object('celery_tasks.config')
# 注册任务
celery_app.autodiscover_tasks(['celery_tasks.email']) # 注意:传递的参数是任务列表

分析一下这个程序:

  • "from celery import celery"是导入celery中的celery类。celery_app
  • celery_app是celery类的实例。
  • 把celery配置存放进project/config.py文件,使用celery_app.config_from_object加载配置。
  • 将任务注册到应用中

5、接着是配置文件config.py,代码如下:

broker_url = 'redis://localhost:6379/1' # 使用redis作为消息代理

celery_result_backend = 'redis://localhost:6379/0' # 把任务结果存在了redis

# celery_task_serializer = 'msgpack' # 任务序列化和反序列化使用msgpack方案

celery_result_serializer = 'json' # 读取任务结果一般性能要求不高,所以使用了可读性更好的json

celery_task_result_expires = 60 * 60 * 24 # celery任务结果有效期

celery_accept_content = ['json', 'msgpack'] # 指定接受的内容类型

celery_timezone = 'asia/shanghai'       # celery使用的时区
celery_enable_utc = true            # 启动时区设置
celeryd_log_file = "/var/log/celery/celery.log"   # celery日志存储位置

6、创建email目录,目录下创建tesks.py文件用来编写发送邮件的代码,代码如下:

import time
from celery_tasks.main import celery_app
@celery_app.task(name='seed_email')   # 添加celery_app.task这个装饰器,指定该任务的任务名name='seed_email'
def seed():
  time.sleep(1)
  return "我将发送邮件"

7、在项目app.py中,采用delay()用来调用任务。

from celery_tasks.email.tasks import seed
seed.delay()
seed.delay()
seed.delay()
seed.delay()
seed.delay()

8、项目运行

  首先,我们需要启动redis。接着,切换至proj项目所在目录,并运行命令:

celery -a celery_tasks.main worker -l info

  界面如下:

Python celery原理及运行流程解析

然后,我们运行app.py,app.py调用添加异步任务,输出的结果如下:

Python celery原理及运行流程解析

Python celery原理及运行流程解析

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。

相关标签: Python celery