在程序的運行過程中,我們經(jīng)常會碰到一些耗時耗資源的操作,為了避免它們阻塞主程序的運行,我們經(jīng)常會采用多線程或異步任務(wù)。比如,在 Web 開發(fā)中,對新用戶的注冊,我們通常會給他發(fā)一封激活郵件,而發(fā)郵件是個 IO 阻塞式任務(wù),如果直接把它放到應(yīng)用當(dāng)中,就需要等郵件發(fā)出去之后才能進行下一步操作,此時用戶只能等待再等待。更好的方式是在業(yè)務(wù)邏輯中觸發(fā)一個發(fā)郵件的異步任務(wù),而主程序可以繼續(xù)往下運行。
Celery 是一個強大的分布式任務(wù)隊列,它可以讓任務(wù)的執(zhí)行完全脫離主程序,甚至可以被分配到其他主機上運行。我們通常使用它來實現(xiàn)異步任務(wù)(async task)和定時任務(wù)(crontab)。它的架構(gòu)組成如下圖:
可以看到,Celery 主要包含以下幾個模塊:
任務(wù)模塊
包含異步任務(wù)和定時任務(wù)。其中,異步任務(wù)通常在業(yè)務(wù)邏輯中被觸發(fā)并發(fā)往任務(wù)隊列,而定時任務(wù)由 Celery Beat 進程周期性地將任務(wù)發(fā)往任務(wù)隊列。
消息中間件 Broker
Broker,即為任務(wù)調(diào)度隊列,接收任務(wù)生產(chǎn)者發(fā)來的消息(即任務(wù)),將任務(wù)存入隊列。Celery 本身不提供隊列服務(wù),官方推薦使用 RabbitMQ 和 Redis 等。
任務(wù)執(zhí)行單元 Worker
Worker 是執(zhí)行任務(wù)的處理單元,它實時監(jiān)控消息隊列,獲取隊列中調(diào)度的任務(wù),并執(zhí)行它。
任務(wù)結(jié)果存儲 Backend
Backend 用于存儲任務(wù)的執(zhí)行結(jié)果,以供查詢。同消息中間件一樣,存儲也可使用 RabbitMQ, Redis 和 MongoDB 等。
使用 Celery 實現(xiàn)異步任務(wù)主要包含三個步驟:
為了簡單起見,對于 Broker 和 Backend,這里都使用 redis。在運行下面的例子之前,請確保 redis 已正確安裝,并開啟 redis 服務(wù),當(dāng)然,celery 也是要安裝的??梢允褂孟旅娴拿顏戆惭b celery 及相關(guān)依賴:
$ pip install 'celery[redis]'
將下面的代碼保存為文件 tasks.py
:
# -*- coding: utf-8 -*-
import time
from celery import Celery
broker = 'redis://127.0.0.1:6379'
backend = 'redis://127.0.0.1:6379/0'
app = Celery('my_task', broker=broker, backend=backend)
@app.task
def add(x, y):
time.sleep(5) # 模擬耗時操作
return x + y
上面的代碼做了幾件事:
my_task
;redis://127.0.0.1:6379
;redis://127.0.0.1:6379/0
;add
,當(dāng)函數(shù)被 @app.task
裝飾后,就成為可被 Celery 調(diào)度的任務(wù);在當(dāng)前目錄,使用如下方式啟動 Celery Worker:
$ celery worker -A tasks --loglevel=info
其中:
-A
指定了 Celery 實例的位置,本例是在 tasks.py
中,Celery 會自動在該文件中尋找 Celery 對象實例,當(dāng)然,我們也可以自己指定,在本例,使用 -A tasks.app
;--loglevel
指定了日志級別,默認為 warning,也可以使用 -l info
來表示;在生產(chǎn)環(huán)境中,我們通常會使用 Supervisor 來控制 Celery Worker 進程。
啟動成功后,控制臺會顯示如下輸出:
現(xiàn)在,我們可以在應(yīng)用程序中使用 delay()
或 apply_async()
方法來調(diào)用任務(wù)。
在當(dāng)前目錄打開 Python 控制臺,輸入以下代碼:
>>> from tasks import add
>>> add.delay(2, 8)
<AsyncResult: 2272ddce-8be5-493f-b5ff-35a0d9fe600f>
在上面,我們從 tasks.py
文件中導(dǎo)入了 add
任務(wù)對象,然后使用 delay()
方法將任務(wù)發(fā)送到消息中間件(Broker),Celery Worker 進程監(jiān)控到該任務(wù)后,就會進行執(zhí)行。我們將窗口切換到 Worker 的啟動窗口,會看到多了兩條日志:
[2016-12-10 12:00:50,376: INFO/MainProcess] Received task: tasks.add[2272ddce-8be5-493f-b5ff-35a0d9fe600f]
[2016-12-10 12:00:55,385: INFO/PoolWorker-4] Task tasks.add[2272ddce-8be5-493f-b5ff-35a0d9fe600f] succeeded in 5.00642602402s: 10
這說明任務(wù)已經(jīng)被調(diào)度并執(zhí)行成功。
另外,我們?nèi)绻氆@取執(zhí)行后的結(jié)果,可以這樣做:
>>> result = add.delay(2, 6)
>>> result.ready() # 使用 ready() 判斷任務(wù)是否執(zhí)行完畢
False
>>> result.ready()
False
>>> result.ready()
True
>>> result.get() # 使用 get() 獲取任務(wù)結(jié)果
8
在上面,我們是在 Python 的環(huán)境中調(diào)用任務(wù)。事實上,我們通常在應(yīng)用程序中調(diào)用任務(wù)。比如,將下面的代碼保存為 client.py
:
# -*- coding: utf-8 -*-
from tasks import add
# 異步任務(wù)
add.delay(2, 8)
print 'hello world'
運行命令 $ python client.py
,可以看到,雖然任務(wù)函數(shù) add
需要等待 5 秒才返回執(zhí)行結(jié)果,但由于它是一個異步任務(wù),不會阻塞當(dāng)前的主程序,因此主程序會往下執(zhí)行 print
語句,打印出結(jié)果。
在上面的例子中,我們直接把 Broker 和 Backend 的配置寫在了程序當(dāng)中,更好的做法是將配置項統(tǒng)一寫入到一個配置文件中,通常我們將該文件命名為 celeryconfig.py
。Celery 的配置比較多,可以在官方文檔查詢每個配置項的含義。
下面,我們再看一個例子。項目結(jié)構(gòu)如下:
celery_demo # 項目根目錄
├── celery_app # 存放 celery 相關(guān)文件
│?? ├── __init__.py
│?? ├── celeryconfig.py # 配置文件
│?? ├── task1.py # 任務(wù)文件 1
│?? └── task2.py # 任務(wù)文件 2
└── client.py # 應(yīng)用程序
__init__.py
代碼如下:
# -*- coding: utf-8 -*-
from celery import Celery
app = Celery('demo') # 創(chuàng)建 Celery 實例
app.config_from_object('celery_app.celeryconfig') # 通過 Celery 實例加載配置模塊
celeryconfig.py
代碼如下:
BROKER_URL = 'redis://127.0.0.1:6379' # 指定 Broker
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0' # 指定 Backend
CELERY_TIMEZONE='Asia/Shanghai' # 指定時區(qū),默認是 UTC
# CELERY_TIMEZONE='UTC'
CELERY_IMPORTS = ( # 指定導(dǎo)入的任務(wù)模塊
'celery_app.task1',
'celery_app.task2'
)
task1.py
代碼如下:
import time
from celery_app import app
@app.task
def add(x, y):
time.sleep(2)
return x + y
task2.py
代碼如下:
import time
from celery_app import app
@app.task
def multiply(x, y):
time.sleep(2)
return x * y
client.py
代碼如下:
# -*- coding: utf-8 -*-
from celery_app import task1
from celery_app import task2
task1.add.apply_async(args=[2, 8]) # 也可用 task1.add.delay(2, 8)
task2.multiply.apply_async(args=[3, 7]) # 也可用 task2.multiply.delay(3, 7)
print 'hello world'
現(xiàn)在,讓我們啟動 Celery Worker 進程,在項目的根目錄下執(zhí)行下面命令:
celery_demo $ celery -A celery_app worker --loglevel=info
接著,運行 $ python client.py
,它會發(fā)送兩個異步任務(wù)到 Broker,在 Worker 的窗口我們可以看到如下輸出:
[2016-12-10 13:51:58,939: INFO/MainProcess] Received task: celery_app.task1.add[9ccffad0-aca4-4875-84ce-0ccfce5a83aa]
[2016-12-10 13:51:58,941: INFO/MainProcess] Received task: celery_app.task2.multiply[64b1f889-c892-4333-bd1d-ac667e677a8a]
[2016-12-10 13:52:00,948: INFO/PoolWorker-3] Task celery_app.task1.add[9ccffad0-aca4-4875-84ce-0ccfce5a83aa] succeeded in 2.00600231002s: 10
[2016-12-10 13:52:00,949: INFO/PoolWorker-4] Task celery_app.task2.multiply[64b1f889-c892-4333-bd1d-ac667e677a8a] succeeded in 2.00601326401s: 21
在前面的例子中,我們使用 delay()
或 apply_async()
方法來調(diào)用任務(wù)。事實上,delay
方法封裝了 apply_async
,如下:
def delay(self, *partial_args, **partial_kwargs):
"""Shortcut to :meth:`apply_async` using star arguments."""
return self.apply_async(partial_args, partial_kwargs)
也就是說,delay
是使用 apply_async
的快捷方式。apply_async
支持更多的參數(shù),它的一般形式如下:
apply_async(args=(), kwargs={}, route_name=None, **options)
apply_async 常用的參數(shù)如下:
task1.apply_async(args=(2, 3), countdown=5) # 5 秒后執(zhí)行任務(wù)
from datetime import datetime, timedelta
# 當(dāng)前 UTC 時間再加 10 秒后執(zhí)行任務(wù)
task1.multiply.apply_async(args=[3, 7], eta=datetime.utcnow() + timedelta(seconds=10))
task1.multiply.apply_async(args=[3, 7], expires=10) # 10 秒后過期
更多的參數(shù)列表可以在官方文檔中查看。
Celery 除了可以執(zhí)行異步任務(wù),也支持執(zhí)行周期性任務(wù)(Periodic Tasks),或者說定時任務(wù)。Celery Beat 進程通過讀取配置文件的內(nèi)容,周期性地將定時任務(wù)發(fā)往任務(wù)隊列。
讓我們看看例子,項目結(jié)構(gòu)如下:
celery_demo # 項目根目錄
├── celery_app # 存放 celery 相關(guān)文件
?? ├── __init__.py
?? ├── celeryconfig.py # 配置文件
?? ├── task1.py # 任務(wù)文件
?? └── task2.py # 任務(wù)文件
__init__.py
代碼如下:
# -*- coding: utf-8 -*-
from celery import Celery
app = Celery('demo')
app.config_from_object('celery_app.celeryconfig')
celeryconfig.py
代碼如下:
# -*- coding: utf-8 -*-
from datetime import timedelta
from celery.schedules import crontab
# Broker and Backend
BROKER_URL = 'redis://127.0.0.1:6379'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'
# Timezone
CELERY_TIMEZONE='Asia/Shanghai' # 指定時區(qū),不指定默認為 'UTC'
# CELERY_TIMEZONE='UTC'
# import
CELERY_IMPORTS = (
'celery_app.task1',
'celery_app.task2'
)
# schedules
CELERYBEAT_SCHEDULE = {
'add-every-30-seconds': {
'task': 'celery_app.task1.add',
'schedule': timedelta(seconds=30), # 每 30 秒執(zhí)行一次
'args': (5, 8) # 任務(wù)函數(shù)參數(shù)
},
'multiply-at-some-time': {
'task': 'celery_app.task2.multiply',
'schedule': crontab(hour=9, minute=50), # 每天早上 9 點 50 分執(zhí)行一次
'args': (3, 7) # 任務(wù)函數(shù)參數(shù)
}
}
task1.py
代碼如下:
import time
from celery_app import app
@app.task
def add(x, y):
time.sleep(2)
return x + y
task2.py
代碼如下:
import time
from celery_app import app
@app.task
def multiply(x, y):
time.sleep(2)
return x * y
現(xiàn)在,讓我們啟動 Celery Worker 進程,在項目的根目錄下執(zhí)行下面命令:
celery_demo $ celery -A celery_app worker --loglevel=info
接著,啟動 Celery Beat 進程,定時將任務(wù)發(fā)送到 Broker,在項目根目錄下執(zhí)行下面命令:
celery_demo $ celery beat -A celery_app
celery beat v4.0.1 (latentcall) is starting.
__ - ... __ - _
LocalTime -> 2016-12-11 09:48:16
Configuration ->
. broker -> redis://127.0.0.1:6379//
. loader -> celery.loaders.app.AppLoader
. scheduler -> celery.beat.PersistentScheduler
. db -> celerybeat-schedule
. logfile -> [stderr]@%WARNING
. maxinterval -> 5.00 minutes (300s)
之后,在 Worker 窗口我們可以看到,任務(wù) task1
每 30 秒執(zhí)行一次,而 task2
每天早上 9 點 50 分執(zhí)行一次。
在上面,我們用兩個命令啟動了 Worker 進程和 Beat 進程,我們也可以將它們放在一個命令中:
$ celery -B -A celery_app worker --loglevel=info
Celery 周期性任務(wù)也有多個配置項,可參考官方文檔。