Python celery原理及运行流程解析

Dear 丶 2023-02-18 08:24 28阅读 0赞

更多python视频教程请到菜鸟教程https://www.piaodoo.com/

**celery简介**

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

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

![202006131007547.png][]

组件介绍：

*  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异步任务列表。如图：

![202006131007558.png][]

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

界面如下：

![202006131007559.png][]

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

![2020061310075510.png][]

![2020061310075511.png][]

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

[Eventlet]: http://eventlet.net/
[gevent]: http://gevent.org/
[202006131007547.png]: https://img.jbzj.com/file_images/article/202006/202006131007547.png
[202006131007558.png]: https://img.jbzj.com/file_images/article/202006/202006131007558.png
[202006131007559.png]: https://img.jbzj.com/file_images/article/202006/202006131007559.png
[2020061310075510.png]: https://img.jbzj.com/file_images/article/202006/2020061310075510.png
[2020061310075511.png]: https://img.jbzj.com/file_images/article/202006/2020061310075511.png