Python Celery异步任务队列原理与实战部署指南

Celery简介

Celery简介

Celery是一个基于分布式消息传输机制的异步任务队列系统，专注于实时处理场景，同时具备任务调度能力。其基本执行单元为任务（task），支持通过多线程或协程库（如Eventlet、gevent等）在单个或多个Worker服务器上进行并发执行。任务既可以在后台异步运行，也可以同步等待执行结果返回。

在生产环境中，Celery能够高效处理日均百万级的任务量。其完整的系统架构如下图所示：

核心组件介绍

核心组件介绍如下：

• Producer（生产者）：通过调用Celery提供的API、函数或装饰器生成任务，并将任务提交给任务队列进行处理的角色。
• Celery Beat：作为任务调度器，Beat进程会定期读取配置文件，将配置中设定周期执行的任务按时发送给任务队列。
• Broker（消息代理）：即消息中间件，负责接收Producer发送的任务消息，将其存入队列后按顺序分发给Consumer（消费者）。Celery支持多种后端存储，包括RabbitMQ、Redis、MongoDB、Beanstalk、SQLAlchemy和Zookeeper等。尽管支持多种选项，但仅RabbitMQ和Redis适用于生产环境，官方强烈推荐使用RabbitMQ。
• Celery Worker：作为任务的消费者，通常会在多台服务器上部署多个Worker实例以提升整体执行效率。
• Result Backend（结果后端）：用于保存任务执行后的状态信息和最终结果，供用户查询。Celery默认支持Redis、RabbitMQ、MongoDB、Django ORM和SQLAlchemy等多种存储方式。

工作原理

工作原理

Celery的核心职责是管理任务在不同服务器间的分配并获取执行结果。至于服务器间如何通信，Celery本身并不直接解决，而是依赖外部消息队列工具。因此，RabbitMQ或Redis等消息队列管理工具被引入并与Celery集成，负责处理服务器间的通信任务。Celery本身不具备消息存储功能，必须依赖介质（如RabbitMQ、Redis、MySQL或MongoDB）进行存储。鉴于RabbitMQ在速度和可用性方面的优势，官方推荐使用它作为消息代理。

Celery安装及使用

Celery安装及使用

1、安装Celery

pip install celery

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

celery worker --help

3、创建一个工程项目project，然后在项目内创建一个名为celery_tasks的异步任务模块。目录结构如图：

主程序main.py实现

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_app是Celery类的实例化对象。
• 将Celery的配置信息存放于project/config.py文件中，并通过celery_app.config_from_object方法加载配置。
• 将具体任务注册到Celery应用中。

配置文件config.py配置

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日志存储位置

任务定义tasks.py编写

6、创建email目录，并在该目录下创建tasks.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

执行后的界面输出如下：

随后，运行app.py文件。app.py调用添加异步任务后，输出结果如下：

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