Python环境下安装使用异步任务队列包Celery的基础教程-Python Tutorial-php.cn

1.简介

celery(芹菜)是一个异步任务队列/基于分布式消息传递的作业队列。它侧重于实时操作，但对调度支持也很好。
celery用于生产系统每天处理数以百万计的任务。
celery是用Python编写的，但该协议可以在任何语言实现。它也可以与其他语言通过webhooks实现。
建议的消息代理RabbitMQ的，但提供有限支持Redis, Beanstalk, MongoDB, CouchDB, ，和数据库（使用SQLAlchemy的或Django的 ORM）。
celery是易于集成Django, Pylons and Flask，使用 django-celery, celery-pylons and Flask-Celery 附加包即可。

2. 安装
有了上面的概念，需要安装这么几个东西：RabbitMQ、SQLAlchemy、Celery
安装方式也都很简单： RabbitMQ:
mac下：

brew install rabbitmq

Copy after login

linux:

sudo apt-get install rabbitmq-server

Copy after login

剩下两个都是Python的东西了，直接pip安装就好了，对于从来没有安装过MySQL驱动的同学可能需要安装MySQL-python。
安装完成之后，启动服务:

$ rabbitmq-server[回车]

Copy after login

启动后不要关闭窗口, 下面操作新建窗口(Tab)

3. 简单案例
确保你之前的RabbitMQ已经启动。
还是官网的那个例子，在任意目录新建一个tasks.py的文件，内容如下：

from celery import Celery

app = Celery('tasks', broker='amqp://guest@localhost//')

@app.task
def add(x, y):
  return x + y

Copy after login

在同级目录执行:

$ celery -A tasks worker --loglevel=info

Copy after login

该命令的意思是启动一个worker，把tasks中的任务（add(x,y)）把任务放到队列中。
保持窗口打开，新开一个窗口进入交互模式，python或者ipython:

>>> from tasks import add
>>> add.delay(4, 4)

Copy after login

到此为止，你已经可以使用celery执行任务了，上面的python交互模式下简单的调用了add任务，并传递4，4参数。
但此时有一个问题，你突然想知道这个任务的执行结果和状态，到底完了没有。因此就需要设置backend了。
修改之前的tasks.py中的代码为：

# coding:utf-8
import subprocess
from time import sleep

from celery import Celery

backend = 'db+mysql://root:@192.168.0.102/celery'
broker = 'amqp://guest@192.168.0.102:5672'

app = Celery('tasks', backend=backend, broker=broker)


@app.task
def add(x, y):
  sleep(10)
  return x + y


@app.task
def hostname():
  return subprocess.check_output(['hostname'])

Copy after login

除了添加backend之外，上面还添加了一个who的方法用来测试多服务器操作。修改完成之后，还是按照之前的方式启动。
同样进入python的交互模型:

>>> from tasks import add, hostname
>>> r = add.delay(4, 4)
>>> r.ready() # 10s内执行，会输出False，因为add中sleep了10s
>>>
>>> r = hostname.delay()
>>> r.result # 输出你的hostname

Copy after login

4. 测试多服务器
做完上面的测试之后，产生了一个疑惑，Celery叫做分布式任务管理，那它的分布式体现在哪？它的任务都是怎么执行的？在哪个机器上执行的？
在当前服务器上的celery服务不关闭的情况下，按照同样的方式在另外一台服务器上安装Celery，并启动:

$ celery -A tasks worker --loglevel=info

Copy after login

发现前一个服务器的Celery服务中输出你刚启动的服务器的hostname，前提是那台服务器连上了你的rabbitmq。
然后再进入python交互模式:

>>> from tasks import hostname
>>>
>>> for i in range(10):
...   r = hostname.delay()
...   print r.result # 输出你的hostname
>>>

Copy after login

看你输入的内容已经观察两台服务器上你启动celery服务的输出。

5. RabbitMQ远程连接的问题
一开始测试时远程服务器无法连接本地的RabbitMQ服务，后来发现需要设置权限，在/usr/local/etc/rabbitmq/rabbitmq-env.conf这个文件中，修改NODE_IP_ADDRESS=127.0.0.1中的ip为0.0.0.0。

6. 总结的说
这篇文章简单的介绍了Celery的使用，重点还是在分布式的使用。觉得不太爽的地方是，在扩展时，需要重新把代码（tasks.py）部署一遍，而不是可以直接把tasks进行共享，可能Celery是通过task来进行不同的worker的匹配的？目前还不太了解，等深入使用之后再说。

Related labels：

celery python asynchronous queue

source：php.cn

Previous article：Python编程中归并排序算法的实现步骤详解 Next article：Python手机号码归属地查询代码

Statement of this Website

The content of this article is voluntarily contributed by netizens, and the copyright belongs to the original author. This site does not assume corresponding legal responsibility. If you find any content suspected of plagiarism or infringement, please contact admin@php.cn