详解Python Celery和RabbitMQ实战教程

Celery是一个异步任务队列。它可以用于需要异步运行的任何内容。RabbitMQ是Celery广泛使用的消息代理。在本这篇文章中，我将使用RabbitMQ来介绍Celery的基本概念，然后为一个小型演示项目设置Celery 。最后，设置一个Celery Web控制台来监视我的任务

  来！看图说话：

BrokerBroker(RabbitMQ)负责创建任务队列，根据一些路由规则将任务分派到任务队列，然后将任务从任务队列交付给worker

Consumer (Celery Workers)Consumer是执行任务的一个或多个Celery workers。可以根据用例启动许多workers

Result Backend后端用于存储任务的结果。但是，它不是必需的元素，如果不在设置中包含它，就无法访问任务的结果

  首先，需要安装好Celery，可以使用PyPI：

pip install celery选择一个Broker：RabbitMQ

  为什么我们需要broker呢?这是因为Celery本身并不构造消息队列，所以它需要一个额外的消息传输来完成这项工作。这里可以将Celery看作消息代理的包装器

实际上，也可以从几个不同的代理中进行选择，比如RabbitMQ、Redis或数据库(例如Django数据库)

在这里使用RabbitMQ作为代理，因为它功能完整、稳定，Celery推荐使用它。由于演示我的环境是在Mac OS中，安装RabbitMQ使用Homebrew即可：

brew install rabbitmq#如果是Ubuntu的话使用apt-get安装启动RabbitMQ

  程序将在/usr/local/sbin中安装RabbitMQ，虽然有些系统可能会有所不同。可以将此路径添加到环境变量路径，以便以后方便地使用。例如，打开shell启动文件~/.bash_profile添加：

PATH=$PATH:/usr/local/sbin

现在，可以使用rabbitmq-server命令启动我们的RabbitMQ服务器。检查RabbitMQ服务器成功启动，将看到类似的输出：

  RabbitMQ使用Celery之前，需要对RabbitMQ进行一些配置。简单地说，我们需要创建一个虚拟主机和用户，然后设置用户权限，以便它可以访问虚拟主机

# 添加用户跟密码$ 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_importfrom celery import Celeryapp = 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_importfrom test_celery.celery import appimport time@app.taskdef 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_longtimeimport timeif __name__ == ’__main__’: result = add_longtime.delay(1,2)#此时，任务还未完成，它将返回False print ’Task finished? ’, result.ready() print ’Task result: ’, result.result # 延长到10秒以确保任务已经完成 time.sleep(10) # 现在任务完成，ready方法将返回True print ’Task finished? ’, result.ready() print ’Task result: ’, result.result

这里，我们使用delay方法调用任务add_longtime，如果我们想异步处理任务，就需要使用delay方法。此外，保存任务的结果并打印一些信息。如果任务已经完成，ready方法将返回True，否则返回False。result属性是任务的结果，如果任务尚未完成，则返回None。

现在，可以使用下面的命令启动Celery(注：在项目文件夹中运行):

celery -A test_celery worker --loglevel=info

Celery成功连接到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 beginslong time task finished

最后一行显示我们的任务在5秒内完成，任务结果为3:

Task test_celery.tasks.add_longtime[7d942984-8ea6-4e4d-8097-225616f797d5] succeeded in 5.025242167s: 3

在当前控制台中，您将看到以下输出:

Flower是一款基于网络的Celery实时监控软件。使用Flower，可以轻松地监视任务进度和历史记录

使用pip来安装Flower:

pip install flower

要启动Flower web控制台，需要运行以下命令：

celery -A test_celery flower

Flower将运行具有默认端口5555的服务器，可以通过http://localhost:5555访问web控制台

到此这篇关于详解Python Celery和RabbitMQ实战教程的文章就介绍到这了,更多相关Python Celery和RabbitMQ实战内容请搜索好吧啦网以前的文章或继续浏览下面的相关文章希望大家以后多多支持好吧啦网！