详解Python Celery和RabbitMQ实战教程
(编辑：jimmy 日期: 2024/12/27 浏览：2)
前言
Celery是一个异步任务队列。它可以用于需要异步运行的任何内容。RabbitMQ是Celery广泛使用的消息代理。在本这篇文章中，我将使用RabbitMQ来介绍Celery的基本概念，然后为一个小型演示项目设置Celery 。最后，设置一个Celery Web控制台来监视我的任务
基本概念
"text-align: center">
Broker
Broker(RabbitMQ)负责创建任务队列，根据一些路由规则将任务分派到任务队列，然后将任务从任务队列交付给worker
Consumer (Celery Workers)
Consumer是执行任务的一个或多个Celery workers。可以根据用例启动许多workers
Result Backend
后端用于存储任务的结果。但是，它不是必需的元素，如果不在设置中包含它，就无法访问任务的结果
安装Celery
"_blank" href="https://pypi.org/" rel="external nofollow" >PyPI：
pip install celery
选择一个Broker：RabbitMQ
"htmlcode">
brew install rabbitmq #如果是Ubuntu的话使用apt-get安装
启动RabbitMQ
"htmlcode">
PATH=$PATH:/usr/local/sbin
现在，可以使用rabbitmq-server命令启动我们的RabbitMQ服务器。检查RabbitMQ服务器成功启动，将看到类似的输出：
为Celery配置RabbitMQ
"htmlcode">
# 添加用户跟密码 $ rabbitmqctl add_user test test123 # 添加虚拟主机 $ rabbitmqctl add_vhost test_vhost # 为用户添加标签 $ rabbitmqctl set_user_tags test test_tag # 设置用户权限 $ rabbitmqctl set_permissions -p test_vhost test ".*" ".*" ".*"
敲黑板！RabbitMQ中有三种操作:配置、写入和读取
上面命令末尾的字符串表示用户test将拥有所有配置、写入和读取权限
演示项目
现在让我们创建一个简单的项目来演示Celery的使用
在celery.py中添加以下代码:
from __future__ import absolute_import
from celery import Celery
app = Celery('test_celery',
broker='amqp://test:test123@localhost/test_vhost',
backend='rpc://',
include=['test_celery.tasks'])
在这里，初始化了一个名为app的Celery实例，将用于创建一个任务。Celery的第一个参数只是项目包的名称，即“test_celery”。
broker参数指定代理URL，对于RabbitMQ，传输是amqp。
后端参数指定后端URL。Celery中的后端用于存储任务结果。因此，如果需要在任务完成时访问任务的结果，应该为Celery设置一个后端。
rpc意味着将结果作为AMQP消息发送回去，这对本次演示来说是一种可接受的格式
include参数指定了在Celery工作程序启动时要导入的模块列表。我们在这里添加了tasks模块，以便找到我们的任务。
在tasks.py这个文件中，定义了我们的任务add_longtime:
from __future__ import absolute_import from test_celery.celery import app import time @app.task def add_longtime(a, b): print 'long time task begins' # sleep 5 seconds time.sleep(5) print 'long time task finished' return a + b
可以看到，导入了在前面的Celery模块中定义的应用程序，并将其用作任务方法的装饰器。另外注意！app.task只是一个装饰器。此外，我们在add_longtime任务中休眠5秒，以模拟一个耗时较长的Task
在设置好Celery之后，我们需要开始运行任务，它包含在runs_tasks.py:
from .tasks import add_longtime import time if __name__ == '__main__': result = add_longtime.delay(1,2) #此时，任务还未完成，它将返回False print 'Task finished"htmlcode">celery -A test_celery worker --loglevel=infoCelery成功连接到RabbitMQ，你会看到这样的东西:
运行任务
再项目文件中输入以下命令运行它：
python -m test_celery.run_tasks查看Celery控制台，看到运行任务:
[2020-05-15 17:15:21,508: INFO/MainProcess]
Received task: test_celery.tasks.add_longtime[25ba9c87-69a7-4383-b983-1cefdb32f8b3]
[2020-05-15 17:15:21,508: WARNING/Worker-3] long time task begins
[2020-05-15 17:15:31,510: WARNING/Worker-3] long time task finished
[2020-05-15 17:15:31,512: INFO/MainProcess]
Task test_celery.tasks.add_longtime[25ba9c87-69a7-4383-b983-1cefdb32f8b3] succeeded in 15.003732774s: 3当Celery收到一个任务，它打印出任务名称与任务id(在括号中):
Received task: test_celery.tasks.add_longtime[7d942984-8ea6-4e4d-8097-225616f797d5]
在这一行下面是我们的任务add_longtime打印的两行，时间延迟为5秒:
long time task begins long time task finished最后一行显示我们的任务在5秒内完成，任务结果为3:
Task test_celery.tasks.add_longtime[7d942984-8ea6-4e4d-8097-225616f797d5] succeeded in 5.025242167s: 3在当前控制台中，您将看到以下输出:
实时监控Celery
Flower是一款基于网络的Celery实时监控软件。使用Flower，可以轻松地监视任务进度和历史记录
使用pip来安装Flower:
pip install flower要启动Flower web控制台，需要运行以下命令：
celery -A test_celery flowerFlower将运行具有默认端口5555的服务器，可以通过http://localhost:5555访问web控制台
