摘要：我們將窗口切換到的啟動(dòng)窗口，會(huì)看到多了兩條日志這說(shuō)明任務(wù)已經(jīng)被調(diào)度并執(zhí)行成功。本文標(biāo)題為異步任務(wù)神器簡(jiǎn)明筆記本文鏈接為參考資料使用之美分布式任務(wù)隊(duì)列的介紹思誠(chéng)之道異步任務(wù)神器簡(jiǎn)明筆記

在程序的運(yùn)行過(guò)程中，我們經(jīng)常會(huì)碰到一些耗時(shí)耗資源的操作，為了避免它們阻塞主程序的運(yùn)行，我們經(jīng)常會(huì)采用多線程或異步任務(wù)。比如，在 Web 開(kāi)發(fā)中，對(duì)新用戶的注冊(cè)，我們通常會(huì)給他發(fā)一封激活郵件，而發(fā)郵件是個(gè) IO 阻塞式任務(wù)，如果直接把它放到應(yīng)用當(dāng)中，就需要等郵件發(fā)出去之后才能進(jìn)行下一步操作，此時(shí)用戶只能等待再等待。更好的方式是在業(yè)務(wù)邏輯中觸發(fā)一個(gè)發(fā)郵件的異步任務(wù)，而主程序可以繼續(xù)往下運(yùn)行。

Celery 是一個(gè)強(qiáng)大的分布式任務(wù)隊(duì)列，它可以讓任務(wù)的執(zhí)行完全脫離主程序，甚至可以被分配到其他主機(jī)上運(yùn)行。我們通常使用它來(lái)實(shí)現(xiàn)異步任務(wù)（async task）和定時(shí)任務(wù)（crontab）。它的架構(gòu)組成如下圖：

可以看到，Celery 主要包含以下幾個(gè)模塊：

任務(wù)模塊
包含異步任務(wù)和定時(shí)任務(wù)。其中，異步任務(wù)通常在業(yè)務(wù)邏輯中被觸發(fā)并發(fā)往任務(wù)隊(duì)列，而定時(shí)任務(wù)由 Celery Beat 進(jìn)程周期性地將任務(wù)發(fā)往任務(wù)隊(duì)列。

消息中間件 Broker
Broker，即為任務(wù)調(diào)度隊(duì)列，接收任務(wù)生產(chǎn)者發(fā)來(lái)的消息（即任務(wù)），將任務(wù)存入隊(duì)列。Celery 本身不提供隊(duì)列服務(wù)，官方推薦使用 RabbitMQ 和 Redis 等。

任務(wù)執(zhí)行單元 Worker
Worker 是執(zhí)行任務(wù)的處理單元，它實(shí)時(shí)監(jiān)控消息隊(duì)列，獲取隊(duì)列中調(diào)度的任務(wù)，并執(zhí)行它。

任務(wù)結(jié)果存儲(chǔ) Backend
Backend 用于存儲(chǔ)任務(wù)的執(zhí)行結(jié)果，以供查詢。同消息中間件一樣，存儲(chǔ)也可使用 RabbitMQ, Redis 和 MongoDB 等。

使用 Celery 實(shí)現(xiàn)異步任務(wù)主要包含三個(gè)步驟：

創(chuàng)建一個(gè) Celery 實(shí)例

啟動(dòng) Celery Worker

應(yīng)用程序調(diào)用異步任務(wù)

為了簡(jiǎn)單起見(jiàn)，對(duì)于 Broker 和 Backend，這里都使用 redis。在運(yùn)行下面的例子之前，請(qǐng)確保 redis 已正確安裝，并開(kāi)啟 redis 服務(wù)，當(dāng)然，celery 也是要安裝的。可以使用下面的命令來(lái)安裝 celery 及相關(guān)依賴：

$ pip install "celery[redis]"

# -*- 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)     # 模擬耗時(shí)操作
    return x + y

將上面的代碼保存為文件 tasks.py，做了幾件事：

創(chuàng)建了一個(gè) Celery 實(shí)例 app，名稱為 my_task；

指定消息中間件用 redis，URL 為 redis://127.0.0.1:6379；

指定存儲(chǔ)用 redis，URL 為 redis://127.0.0.1:6379/0；

創(chuàng)建了一個(gè) Celery 任務(wù) add，當(dāng)函數(shù)被 @app.task 裝飾后，就成為可被 Celery 調(diào)度的任務(wù)；

在當(dāng)前目錄，使用如下方式啟動(dòng) Celery Worker：

$ celery worker -A tasks --loglevel=info

其中：

參數(shù) -A 指定了 Celery 實(shí)例的位置，本例是在 tasks.py 中，Celery 會(huì)自動(dòng)在該文件中尋找 Celery 對(duì)象實(shí)例，當(dāng)然，我們也可以自己指定，在本例，使用 -A tasks.app；

參數(shù) --loglevel 指定了日志級(jí)別，默認(rèn)為 warning，也可以使用 -l info 來(lái)表示；

在生產(chǎn)環(huán)境中，我們通常會(huì)使用 Supervisor 來(lái)控制 Celery Worker 進(jìn)程。

啟動(dòng)成功后，控制臺(tái)會(huì)顯示如下輸出：

現(xiàn)在，我們可以在應(yīng)用程序中使用 delay() 或 apply_async() 方法來(lái)調(diào)用任務(wù)。

在當(dāng)前目錄打開(kāi) Python 控制臺(tái)，輸入以下代碼：

>>> from tasks import add
>>> add.delay(2, 8)

在上面，我們從 tasks.py 文件中導(dǎo)入了 add 任務(wù)對(duì)象，然后使用 delay() 方法將任務(wù)發(fā)送到消息中間件（Broker），Celery Worker 進(jìn)程監(jiān)控到該任務(wù)后，就會(huì)進(jìn)行執(zhí)行。我們將窗口切換到 Worker 的啟動(dòng)窗口，會(huì)看到多了兩條日志：

[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

這說(shuō)明任務(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ù)。事實(shí)上，我們通常在應(yīng)用程序中調(diào)用任務(wù)。比如，將下面的代碼保存為 client.py:

# -*- coding: utf-8 -*-

from tasks import add

# 異步任務(wù)
add.delay(2, 8)

print "hello world"

運(yùn)行命令 $ python client.py，可以看到，雖然任務(wù)函數(shù) add 需要等待 5 秒才返回執(zhí)行結(jié)果，但由于它是一個(gè)異步任務(wù)，不會(huì)阻塞當(dāng)前的主程序，因此主程序會(huì)往下執(zhí)行 print 語(yǔ)句，打印出結(jié)果。

在上面的例子中，我們直接把 Broker 和 Backend 的配置寫(xiě)在了程序當(dāng)中，更好的做法是將配置項(xiàng)統(tǒng)一寫(xiě)入到一個(gè)配置文件中，通常我們將該文件命名為 celeryconfig.py。Celery 的配置比較多，可以在官方文檔查詢每個(gè)配置項(xiàng)的含義。

下面，我們?cè)倏匆粋€(gè)例子。項(xiàng)目結(jié)構(gòu)如下：

celery_demo                    # 項(xiàng)目根目錄
    ├── 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 實(shí)例
app.config_from_object("celery_app.celeryconfig")   # 通過(guò) Celery 實(shí)例加載配置模塊

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"                     # 指定時(shí)區(qū)，默認(rèn)是 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)在，讓我們啟動(dòng) Celery Worker 進(jìn)程，在項(xiàng)目的根目錄下執(zhí)行下面命令：

celery_demo $ celery -A celery_app worker --loglevel=info

接著，運(yùn)行 $ python client.py，它會(huì)發(fā)送兩個(gè)異步任務(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() 方法來(lái)調(diào)用任務(wù)。事實(shí)上，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)

也就是說(shuō)，delay 是使用 apply_async 的快捷方式。apply_async 支持更多的參數(shù)，它的一般形式如下：

apply_async(args=(), kwargs={}, route_name=None, **options)

apply_async 常用的參數(shù)如下：

countdown：指定多少秒后執(zhí)行任務(wù)

task1.apply_async(args=(2, 3), countdown=5)    # 5 秒后執(zhí)行任務(wù)

eta (estimated time of arrival)：指定任務(wù)被調(diào)度的具體時(shí)間，參數(shù)類型是 datetime

from datetime import datetime, timedelta

# 當(dāng)前 UTC 時(shí)間再加 10 秒后執(zhí)行任務(wù)
task1.multiply.apply_async(args=[3, 7], eta=datetime.utcnow() + timedelta(seconds=10))

expires：任務(wù)過(guò)期時(shí)間，參數(shù)類型可以是 int，也可以是 datetime

task1.multiply.apply_async(args=[3, 7], expires=10)    # 10 秒后過(guò)期

更多的參數(shù)列表可以在官方文檔中查看。

Celery 除了可以執(zhí)行異步任務(wù)，也支持執(zhí)行周期性任務(wù)（Periodic Tasks），或者說(shuō)定時(shí)任務(wù)。Celery Beat 進(jìn)程通過(guò)讀取配置文件的內(nèi)容，周期性地將定時(shí)任務(wù)發(fā)往任務(wù)隊(duì)列。

讓我們看看例子，項(xiàng)目結(jié)構(gòu)如下：

celery_demo                    # 項(xiàng)目根目錄
    ├── 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"    # 指定時(shí)區(qū)，不指定默認(rèn)為 "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 點(diǎn) 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)在，讓我們啟動(dòng) Celery Worker 進(jìn)程，在項(xiàng)目的根目錄下執(zhí)行下面命令：

celery_demo $ celery -A celery_app worker --loglevel=info

接著，啟動(dòng) Celery Beat 進(jìn)程，定時(shí)將任務(wù)發(fā)送到 Broker，在項(xiàng)目根目錄下執(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 點(diǎn) 50 分執(zhí)行一次。

在上面，我們用兩個(gè)命令啟動(dòng)了 Worker 進(jìn)程和 Beat 進(jìn)程，我們也可以將它們放在一個(gè)命令中：

$ celery -B -A celery_app worker --loglevel=info

Celery 周期性任務(wù)也有多個(gè)配置項(xiàng)，可參考官方文檔。

本文由 funhacks 發(fā)表于個(gè)人博客，采用 Creative Commons BY-NC-ND 4.0（自由轉(zhuǎn)載-保持署名-非商用-禁止演繹）協(xié)議發(fā)布。
非商業(yè)轉(zhuǎn)載請(qǐng)注明作者及出處。商業(yè)轉(zhuǎn)載請(qǐng)聯(lián)系作者本人。
本文標(biāo)題為: 異步任務(wù)神器 Celery 簡(jiǎn)明筆記
本文鏈接為: https://funhacks.net/2016/12/...

Celery - Distributed Task Queue — Celery 4.0.1 documentation

使用Celery - Python之美

分布式任務(wù)隊(duì)列Celery的介紹 – 思誠(chéng)之道

異步任務(wù)神器 Celery 簡(jiǎn)明筆記