Celery
在程式的運作過程中，我們經常會碰到一些耗時耗資源的操作，為了避免它們阻塞主程式的運作，我們經常會采用多線程或異步任務。比如，在 Web 開發中，對新使用者的注冊，我們通常會給他發一封激活郵件，而發郵件是個 IO 阻塞式任務，如果直接把它放到應用當中，就需要等郵件發出去之後才能進行下一步操作，此時使用者隻能等待再等待。更好的方式是在業務邏輯中觸發一個發郵件的異步任務，而主程式可以繼續往下運作。
Celery 是一個強大的分布式任務隊列，它可以讓任務的執行完全脫離主程式，甚至可以被配置設定到其他主機上運作。我們通常使用它來實作異步任務（async task）和定時任務（crontab）。它的架構組成如下圖：
Celery_framework
可以看到，Celery 主要包含以下幾個子產品：
-
任務子產品 Task
包含異步任務和定時任務。其中，異步任務通常在業務邏輯中被觸發并發往任務隊列，而定時任務由 Celery Beat 程序周期性地将任務發往任務隊列。
-
消息中間件 Broker
Broker，即為任務排程隊列，接收任務生産者發來的消息（即任務），将任務存入隊列。Celery 本身不提供隊列服務，官方推薦使用 RabbitMQ 和 Redis 等。
-
任務執行單元 Worker
Worker 是執行任務的處理單元，它實時監控消息隊列，擷取隊列中排程的任務，并執行它。
-
任務結果存儲 Backend
Backend 用于存儲任務的執行結果，以供查詢。同消息中間件一樣，存儲也可使用 RabbitMQ, Redis 和 MongoDB 等。
異步任務
使用 Celery 實作異步任務主要包含三個步驟：
- 建立一個 Celery 執行個體
- 啟動 Celery Worker
- 應用程式調用異步任務
快速入門
為了簡單起見，對于 Broker 和 Backend，這裡都使用 redis。在運作下面的例子之前，請確定 redis 已正确安裝，并開啟 redis 服務，當然，celery 也是要安裝的。可以使用下面的指令來安裝 celery 及相關依賴：
| 1 | $ pip install 'celery[redis]' |
建立 Celery 執行個體
将下面的代碼儲存為檔案
tasks.py
：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 | # -*- 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 |
上面的代碼做了幾件事：
- 建立了一個 Celery 執行個體 app，名稱為
my_task
- 指定消息中間件用 redis，URL 為
redis://127.0.0.1:6379
- 指定存儲用 redis，URL 為
redis://127.0.0.1:6379/0
- 建立了一個 Celery 任務
add
@app.task
啟動 Celery Worker
在目前目錄，使用如下方式啟動 Celery Worker：
| 1 | $ celery worker -A tasks --loglevel=info |
其中：
- 參數
-A
tasks.py
-A tasks.app
- 參數
--loglevel
-l info
在生産環境中，我們通常會使用 Supervisor 來控制 Celery Worker 程序。
啟動成功後，控制台會顯示如下輸出：
celery
調用任務
現在，我們可以在應用程式中使用
delay()
或
apply_async()
方法來調用任務。
在目前目錄打開 Python 控制台，輸入以下代碼：
| 1 2 3 | >>> from tasks import add >>> add.delay( 2, 8) <AsyncResult: 2272ddce -8be5 -493f-b5ff -35a0d9fe600f> |
在上面，我們從
tasks.py
檔案中導入了
add
任務對象，然後使用
delay()
方法将任務發送到消息中間件（Broker），Celery Worker 程序監控到該任務後，就會進行執行。我們将視窗切換到 Worker 的啟動視窗，會看到多了兩條日志：
| 1 2 | [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 |
這說明任務已經被排程并執行成功。
另外，我們如果想擷取執行後的結果，可以這樣做：
| 1 2 3 4 5 6 7 8 9 | >>> result = add.delay( 2, 6) >>> result.ready() # 使用 ready() 判斷任務是否執行完畢 False >>> result.ready() False >>> result.ready() True >>> result.get() # 使用 get() 擷取任務結果 8 |
在上面，我們是在 Python 的環境中調用任務。事實上，我們通常在應用程式中調用任務。比如，将下面的代碼儲存為
client.py
:
| 1 2 3 4 5 6 7 8 | # -*- coding: utf-8 -*- from tasks import add # 異步任務 add.delay( 2, 8) print 'hello world' |
運作指令
$ python client.py
，可以看到，雖然任務函數
add
需要等待 5 秒才傳回執行結果，但由于它是一個異步任務，不會阻塞目前的主程式，是以主程式會往下執行
print
語句，列印出結果。
使用配置
在上面的例子中，我們直接把 Broker 和 Backend 的配置寫在了程式當中，更好的做法是将配置項統一寫入到一個配置檔案中，通常我們将該檔案命名為
celeryconfig.py
。Celery 的配置比較多，可以在官方文檔查詢每個配置項的含義。
下面，我們再看一個例子。項目結構如下：
| 1 2 3 4 5 6 7 | celery_demo # 項目根目錄 ├── celery_app # 存放 celery 相關檔案 │ ├── __init__.py │ ├── celeryconfig.py # 配置檔案 │ ├── task1.py # 任務檔案 1 │ └── task2.py # 任務檔案 2 └── client.py # 應用程式 |
__init__.py
代碼如下：
| 1 2 3 4 5 6 | # -*- coding: utf-8 -*- from celery import Celery app = Celery( 'demo') # 建立 Celery 執行個體 app.config_from_object( 'celery_app.celeryconfig') # 通過 Celery 執行個體加載配置子產品 |
celeryconfig.py
代碼如下：
| 1 2 3 4 5 6 7 8 9 10 | BROKER_URL = 'redis://127.0.0.1:6379' # 指定 Broker CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0' # 指定 Backend CELERY_TIMEZONE= 'Asia/Shanghai' # 指定時區，預設是 UTC # CELERY_TIMEZONE='UTC' CELERY_IMPORTS = ( # 指定導入的任務子產品 'celery_app.task1', 'celery_app.task2' ) |
task1.py
代碼如下：
| 1 2 3 4 5 6 7 | import time from celery_app import app @app.task def add(x, y): time.sleep( 2) return x + y |
task2.py
代碼如下：
| 1 2 3 4 5 6 7 | import time from celery_app import app @app.task def multiply(x, y): time.sleep( 2) return x * y |
client.py
代碼如下：
| 1 2 3 4 5 6 7 8 9 | # -*- 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' |
現在，讓我們啟動 Celery Worker 程序，在項目的根目錄下執行下面指令：
| 1 | celery_demo $ celery -A celery_app worker --loglevel=info |
接着，運作
$ python client.py
，它會發送兩個異步任務到 Broker，在 Worker 的視窗我們可以看到如下輸出：
| 1 2 3 4 | [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
在前面的例子中，我們使用
delay()
或
apply_async()
方法來調用任務。事實上，
delay
方法封裝了
apply_async
，如下：
| 1 2 3 | 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
支援更多的參數，它的一般形式如下：
| 1 | apply_async(args=(), kwargs={}, route_name= None, **options) |
apply_async 常用的參數如下：
- countdown：指定多少秒後執行任務
| 1 | task1.apply_async(args=(2, 3), countdown=5) # 5 秒後執行任務 |
- eta (estimated time of arrival)：指定任務被排程的具體時間，參數類型是 datetime
| 1 2 3 4 | from datetime import datetime, timedelta # 目前 UTC 時間再加 10 秒後執行任務 task1.multiply.apply_async(args=[ 3, 7], eta=datetime.utcnow() + timedelta(seconds= 10)) |
- expires：任務過期時間，參數類型可以是 int，也可以是 datetime
| 1 | task1.multiply.apply_async(args=[ 3, 7], expires= 10) # 10 秒後過期 |
更多的參數清單可以在官方文檔中檢視。
定時任務
Celery 除了可以執行異步任務，也支援執行周期性任務（Periodic Tasks），或者說定時任務。Celery Beat 程序通過讀取配置檔案的内容，周期性地将定時任務發往任務隊列。
讓我們看看例子，項目結構如下：
| 1 2 3 4 5 6 | celery_demo # 項目根目錄 ├── celery_app # 存放 celery 相關檔案 ├── __init__.py ├── celeryconfig.py # 配置檔案 ├── task1.py # 任務檔案 └── task2.py # 任務檔案 |
__init__.py
代碼如下：
| 1 2 3 4 5 6 | # -*- coding: utf-8 -*- from celery import Celery app = Celery( 'demo') app.config_from_object( 'celery_app.celeryconfig') |
celeryconfig.py
代碼如下：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 | # -*- 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' # 指定時區，不指定預設為 '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 秒執行一次 'args': ( 5, 8) # 任務函數參數 }, 'multiply-at-some-time': { 'task': 'celery_app.task2.multiply', 'schedule': crontab(hour= 9, minute= 50), # 每天早上 9 點 50 分執行一次 'args': ( 3, 7) # 任務函數參數 } } |
task1.py
代碼如下：
| 1 2 3 4 5 6 7 | import time from celery_app import app @app.task def add(x, y): time.sleep( 2) return x + y |
task2.py
代碼如下：
| 1 2 3 4 5 6 7 | import time from celery_app import app @app.task def multiply(x, y): time.sleep( 2) return x * y |
現在，讓我們啟動 Celery Worker 程序，在項目的根目錄下執行下面指令：
| 1 | celery_demo $ celery -A celery_app worker --loglevel=info |
接着，啟動 Celery Beat 程序，定時将任務發送到 Broker，在項目根目錄下執行下面指令：
| 1 2 3 4 5 6 7 8 9 10 11 | 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 視窗我們可以看到，任務
task1
每 30 秒執行一次，而
task2
每天早上 9 點 50 分執行一次。
在上面，我們用兩個指令啟動了 Worker 程序和 Beat 程序，我們也可以将它們放在一個指令中：
| 1 | $ celery -B -A celery_app worker --loglevel=info |
Celery 周期性任務也有多個配置項，可參考官方文檔。
參考資料
- Celery - Distributed Task Queue — Celery 4.0.1 documentation
- 使用Celery - Python之美
- 分布式任務隊列Celery的介紹 – 思誠之道
- 異步任務神器 Celery 簡明筆記