getting started with celery and redis - 18965050/python-celery GitHub Wiki

前言

这是篇翻译稿(http://agiliq.com/blog/2015/07/getting-started-with-celery-and-redis/) 对celery的介绍比较浅显易懂, 因此就拿来翻译了下. 文档示例已全部放到 https://github.com/18965050/python-celery/tree/master/tutorial 目录下

议程

何时使用celery
为什么使用celery
一个简单的celery例子
不使用celery的慢速脚本和使用celery的快速脚本
分离celery配置和代码
多模块中使用celery任务
celery结合package使用
分离celery和redis
分离celery和web应用脚本

何时使用celery

Celery是个任务处理系统, 应用在很多web应用中. 其一般使用在如下场景:

用于请求-响应循环中的网络调用. server需要对其收到的任何web请求作出快速响应. 如果在请求-响应环中存在网络调用,则其应该运行在请求-响应环之外. 比如: 当用户注册一个网站时会需要发送一封激活邮件. 发送邮件就是个网络调用可能会消耗2-3秒, 而用户不应该等待这2-3秒的时间. 因此发送激活邮件应独立于请求-响应环之外. 这就可以通过celery实现
将一个由许多个独立小任务组成的大任务分解. 比如: 你想读取一个用户在facebook上的timeline. facebook对于获取不同的数据已经提供了不同的终端. 一个终端用于获取用户timeline的图片, 另一个终端用于获取用户timeline的帖子, 还有个终端用于获取用户timeline的喜好等等. 如果你写个单独的函数串行的访问facebook提供的5个终端, 假设每个终端的平均响应为2秒, 则此函数执行完成需要10秒来. 因此你可以分解工作为5个独立的任务(这很简单, 一会我们就会看到), 让Celery来处理这些任务. Celery可以同时并发的访问这5个终端, 让你在2秒内获取所有终端的响应(当然, 你也可以使用gevent).

为什么使用celery

我们想要web响应变得更快. 比如对于用户注册, server应该立即发送响应, 而实际发送邮件的工作交由celery完成. celery会在后台, 置身于请求-响应环之外来发送激活邮件我们可以利用celery来使脚本运行更快, 更好的利用cpu.在上述的facebook例子中, 使用celery的话,耗时会从10秒变成2秒,cpu的利用率也会更高.
我们可使用celery让任务更加可管理. 在前面的facebook例子中, 如果所有的事情都是在单函数中顺序的执行并且如果在第二个请求连接中发生了错误, 剩下的3个请求就不会执行. 而如果5个请求在不同的进程中执行, 一个进程发生了错误, 不会影响其他进程执行. 因此如果我们正确使用celery, 任务会变得更可管理.

简单celery例子

假设我们有个函数要获取一些url的响应

不使用celery

import requests
import time

def func(urls):
    start = time.time()
    for url in urls:
        resp = requests.get(url)
        print(resp.status_code)
    print("It took", time.time() - start, "seconds")

if __name__ == "__main__":
    func(["https://www.baidu.com", "http://www.taobao.com", "http://www.163.com", "http://www.xyz.cn"])

运行程序:

python no_celery_blog.py

输出为:

(hack)~/Play/Python/hack $ python no_celery_blog.py
200
200
200
200
200
It took 7.58989787102 seconds

使用celery

celery运行的主要组件是celery worker 在上述用户注册的例子中, celery worker干发送邮件的事在上述facebook例子中, celery worker干访问不同url的事类似的, 在上述no_celery_blog.py例子中, celery worker干访问url的事 celery worker和你的应用或脚本在不同的进程中相互独立的运行. 因此你的应用/脚本和celery需要某种通道来相互通信. 这就是该消息队列出场的时候了. 应用代码需要将任务放到某处, 供celery worker访问和执行. 应用代码将任务放到消息队列中, 而celery worker则从消息对象中获取任务并执行. 我们将使用redis作为消息队列. 请确认你已经安装redis并通过redis-server命令启动redis 请确认你已经安装celery 创建文件celery_blog.py:

import requests
from celery import Celery

app = Celery('celery_blog', broker='redis://192.168.51.17:6379/8')

@app.task
def fetch_url(url):
    resp = requests.get(url)
    print(resp.status_code)

def func(urls):
    for url in urls:
        fetch_url.delay(url)

if __name__ == "__main__":
    func(["https://www.baidu.com", "http://www.taobao.com", "http://www.163.com", "http://www.xyz.cn"])

代码说明:

我们需要个celery实例来正确的启动celery, 因此我们创建个实例,并称之为app 相关celery的文档[在此](http://celery.readthedocs.io/en/latest/getting-started/first-steps-with-celery.html#application celery文档)

Celery的第一个参数为当前模块的名称, 这个参数是必须的, 这样名称才会自动生成, 第二个参数为broker的kv参数,用于指定说使用broker的URL

一个celery worker可运行在多个并行的进程中. 我们希望所有访问的url都并行而不是串行运行, 因此我们需要一个函数来访问一个单独的url, 并并行的运行这5个函数. 因此我们写了个celery任务(fetch_url)来处理单个url访问. 一个celery任务就是一个函数,并且上面添加app.task装饰器. 在上述代码中, 我们创建了5个任务, 每个任务传入一个不同的url 当我们调用fetch_url.delay(url)时, 代码被序列化并放入到消息队列中,也就是redis中. 正在运行的celery worker会从消息队列中读取这些序列化的信息, 反序列化并执行

运行

在第一个终端中, 通过执行redis-server启动redis
在第二个终端中, 通过celery worker -A celery_blog -l info -c 5命令运行celery worker.当看到输出时, 也就说明celery正在运行
在第三个终端中, 运行脚本python celery_blog.py

不同于执行no_celery_blog.py脚本, 你在执行python celery_blog.py终端上看不到输出.这是由于实际的工作不再由你的脚本执行, 转而由celery去执行了. 切换到celery worker运行的终端, 你可以看到如下的输出:

[2015-07-05 12:57:44,705: INFO/Worker-2] Starting new HTTPS connection (1): facebook.com
[2015-07-05 12:57:44,711: INFO/Worker-4] Starting new HTTPS connection (1): twitter.com
[2015-07-05 12:57:44,716: INFO/Worker-3] Starting new HTTPS connection (1): alexa.com
[2015-07-05 12:57:44,791: INFO/Worker-1] Starting new HTTP connection (1): www.google.co.in

[2015-07-05 12:57:45,063: WARNING/Worker-1] 200

[2015-07-05 12:57:45,376: INFO/Worker-5] Starting new HTTPS connection (1): www.amazon.in

[2015-07-05 12:57:46,179: WARNING/Worker-2] 200

[2015-07-05 12:57:46,185: INFO/MainProcess] Task celery_blog.fetch_url[2809a803-00b2-44c7-85e5-3f6f71d3f5e3] succeeded in 1.48678409203s: None
[2015-07-05 12:57:46,218: INFO/MainProcess] Task celery_blog.fetch_url[9d011563-67f9-4961-a61f-19956bf0cf0a] succeeded in 1.50805259595s: None
....
.....

首先, 我们会注意到整个celery的输出短于8秒. 在前面的例子(no_celery_blog.py)中, 访问5个url耗时8秒, 但现在用时3秒或更短了.

理解 celery worker -A celery_blog -l info -c 5

-c 5: 并发数为5, 也就是说celery会运行5个并行的子进程.每个子进程运行一个独立的任务.
-l info: 日志输出级别
-A celery_blog: celery的配置, celery_blog.py文件模块中包括app和celery worker要关注的任务.

理解输出

Celery worker同时运行在5个子进程中, 分别为Worker-1, Worker-2等等.
任务执行顺序不一定是任务在列表中的顺序
当我们运行python celery_blog.py时, 任务被创建并放入消息队列, 比如redis中
celery worker在其他终端中运行, 和redis交互并从消息队列中取出任务
celery worker反序列化每个任务并在一个子进程中运行
celery worker不会等待第一个子进程任务执行完成后再执行第二个任务
当第一个任务在一个子进程中执行时, celery worker或获取第二个任务, 反序列化并放到第二个子进程中执行
这也就是为什么我们的输出是混合的. 比如4个任务正在执行, 但启动第五个任务前, 我们已经得到了第一个任务的执行结果(200)

分离celery配置和代码

在上面的例子中, 我们只写了一个celery任务. 但你的项目可能会跨多个模块, 并且在不同的模块中有不同的任务.因此我们需要将celery的配置和代码分离.

创建celery_config.py

from celery import Celery

app = Celery('celery_config', broker='redis://192.168.51.17:6379/8', backend='redis://[email protected]:6379/9', include=['celery_blog2'])

创建celery_blog2.py:

import requests
from celery_config import app

@app.task
def fetch_url(url):
    resp = requests.get(url)
    print (resp.status_code)

停止原有的celery work, 并运行celery worker -A celery_config -l info -c 5

创建celery_use.py,并运行之:

from celery_blog2 import fetch_url

def func(urls):
    for url in urls:
        fetch_url.delay(url)

if __name__ == "__main__":
    func(["https://www.baidu.com", "http://www.taobao.com", "http://www.163.com", "http://www.xyz.cn"])

输出为:

[2015-07-05 14:52:02,522: INFO/Worker-1] Starting new HTTPS connection (1): google.com
[2015-07-05 14:52:02,522: INFO/Worker-5] Starting new HTTPS connection (1): facebook.com
[2015-07-05 14:52:03,168: INFO/Worker-1] Starting new HTTPS connection (1): www.google.co.in
[2015-07-05 14:52:03,959: INFO/Worker-5] Starting new HTTPS connection (1): www.facebook.com
[2015-07-05 14:52:03,966: WARNING/Worker-1] 200
[2015-07-05 14:52:03,972: INFO/MainProcess] Task celery_blog.fetch_url[7dbf6870-987b-460e-b5f1-ca17af88bc0a] succeeded in 1.45625397097s: None
[2015-07-05 14:52:04,915: WARNING/Worker-5] 200
[2015-07-05 14:52:04,922: INFO/MainProcess] Task celery_blog.fetch_url[d836a878-823f-4ca2-b918-a6ab0622a157] succeeded in 2.40576425701s: None

添加另一个任务到不同的文件

你还可以添加另一个模块并在此模块中定义任务创建一个模块文件celery_add.py:

from celery_config import app

@app.task
def add(a, b):
    return a + b

修改celery_config.py,包含新的模块:

from celery import Celery

app = Celery('celery_config', broker='redis://192.168.51.17:6379/8', backend='redis://[email protected]:6379/9', include=['celery_blog2','celery_add'])

修改celery_use.py,添加add任务调用:

from celery_blog2 import fetch_url

from celery_add import add

def func(urls):
    for url in urls:
        fetch_url.delay(url)

if __name__ == "__main__":
    func(["https://www.baidu.com", "http://www.taobao.com", "http://www.163.com", "http://www.xyz.cn"])

    result = add.delay(4, 4)
    print(result.ready())
    print(result.get(timeout=2))

celery结合package使用

让我们继续修改celery_config.py.假设包含celery_config.py文件的目录为你项目的根目录和celery_config.py同级, 创建一个包pack.因为这是个python包,别忘了添加pack/__init__.py文件创建文件pack/celery_fetch.py:

import requests
from celery_config import app

@app.task
def fetch_url(url):
    resp = requests.get(url)
    print (resp.status_code)

def func(urls):
    for url in urls:
        fetch_url.delay(url)

修改celery_config.py文件为:

from celery import Celery

app = Celery('celery_config', broker='redis://192.168.51.17:6379/8', backend='redis://[email protected]:6379/9', include=['celery_blog2','celery_add','pack.celery_fetch'])

在celery_config.py同级目录中启动celery worker:

celery worker -A celery_config -l info -c 5

确定你在celery启动输出中看到:

[tasks]
  . pack.celery_fetch.fetch_url

创建celery_use2.py:

from pack.celery_fetch import func

if __name__ == "__main__":
    func(["https://www.baidu.com", "http://www.taobao.com", "http://www.163.com", "http://www.xyz.cn"])

分离celery和redis

到目前为止, 我们的脚本中, celery worker和redis运行在同一台机器上.但实际上这是没有必要的. 它们3个可运行在不同的机器上.
Celery 任务需要进行网络调用. 因此将celery worker放在一台网络状况好的机器上会使得任务运行更快. redis是个内存数据库, 因此大多数情况下,redis会被放在内存服务器上. 在这个例子中, 我们让redis运行在单独的机器上, 而脚本和celery worker运行在本地我让redis运行在54.69.176.94上, 因此需要修改celery_config.py中的broker配置.

app = Celery('celery_config', broker='redis://54.69.176.94:6379/0', include=['celery_blog'])

现在当我运行任务时, 我们的脚本就会被序列化并放到运行在54.69.176.94上的redis中 celery worker也会和54.69.176.94通信, 并从redis中获取任务并执行.

注意: 你要使用你自己的运行redis的服务器地址. 我已经在我的服务器上停掉了redis,因此你连接不了

分离celery和web应用脚本

正如我前面所说的, celery worker和你的程序脚本也运行在相互独立的两个进程中. 我们可以在不同的机器上运行它们. 假定你在54.69.176.94上运行celery,但在本地环境中运行程序脚本.你可以拷贝所有文件(也就是celery_config.py和celery_blog.py)到服务器上, 并运行celery worker -A celery_config -l info 在本机运行的任何任务, 都会被入队到broker所指的队列中. 运行在54.69.176.94上的celery worker也会连接相同的broker,并从此broker中获取任务并执行.

陷阱

在上面最简单的celery例子中(celery_blog.py),修改app名称从celery_blog到celery_blo. 并通过如下命令

celery -A celery_blog worker -l info

运行, 查看celery输出, 会告诉我们任务已经注册为celery_blog.fetch_url

[tasks]
  . celery_blog.fetch_url

再运行脚本celery_blog.py,会报如下错误:

[2015-07-05 16:59:22,956: ERROR/MainProcess] Received unregistered task of type 'celery_blo.fetch_url'.
KeyError: 'celery_blo.fetch_url'

这是由于当我们将任务放入队列时, celery使用的是app名称(也就是celery_blo). 但通过celery worker -A celery_blog命令注册任务时任务使用的是文件模块名称(也就是celery_blog)而不是app的名称