celery是Python開發的分布式異步任務排程系統，Celery支援的消息服務有rmq、redis等

Celery 是一個強大的分布式任務隊列，它可以讓任務的執行完全脫離主程式，甚至可以被配置設定到其他主機上運作。我們通常使用它來實作異步任務（async task）和定時任務（crontab）。它的架構組成如下圖：

Celery 異步分布式 171219 定時任務參考資料

celery由5個主要元件組成:

producer: 任務釋出者, 通過調用API向celery釋出任務的程式

celery beat: 任務排程, 根據配置檔案釋出定時任務

worker: 實際執行任務的程式

broker: 接受任務消息,存入隊列再按順序分發給worker執行

backend: 存儲結果的伺服器

Celery 主要包含以下幾個子產品：

任務子產品

包含異步任務和定時任務。其中，異步任務通常在業務邏輯中被觸發并發往任務隊列，而定時任務由 Celery Beat 程序周期性地将任務發往任務隊列。

消息中間件 Broker

Broker，即為任務排程隊列，接收任務生産者發來的消息（即任務），将任務存入隊列。Celery 本身不提供隊列服務，官方推薦使用 RabbitMQ 和 Redis 等。

任務執行單元 Worker

Worker 是執行任務的處理單元，它實時監控消息隊列，擷取隊列中排程的任務，并執行它。

任務結果存儲 Backend

Backend 用于存儲任務的執行結果，以供查詢。同消息中間件一樣，存儲也可使用 RabbitMQ, Redis 和 MongoDB 等。

異步任務：

使用 Celery 實作異步任務主要包含三個步驟：

建立一個 Celery 執行個體
啟動 Celery Worker
應用程式調用異步任務

在運作下面的例子之前，請確定 redis 已正确安裝，并開啟 redis 服務，當然，celery 也是要安裝的。可以使用下面的指令來安裝 celery 及相關依賴：

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

将上面的代碼儲存為檔案

tasks.py

，做了幾件事：

建立了一個 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 排程的任務；

啟動 Celery Worker

在目前目錄，使用如下方式啟動 Celery Worker：

其中：

參數 -A 指定了 Celery 執行個體的位置，本例是在 tasks.py 中，Celery 會自動在該檔案中尋找 Celery 對象執行個體，當然，我們也可以自己指定，在本例，使用 -A tasks.app ；
參數 --loglevel 指定了日志級别，預設為 warning，也可以使用 -l info 來表示；

在生産環境中，我們通常會使用 Supervisor 來控制 Celery Worker 程序。

調用任務

現在，我們可以在應用程式中使用

delay()

或

apply_async()

方法來調用任務。

在目前目錄打開 Python 控制台，輸入以下代碼：

>>> from tasks import add
>>> add.delay(, )
<AsyncResult: ddcebe5f-b5ffa0d9fe600f>

我們從

tasks.py

檔案中導入了

add

任務對象，然後使用

delay()

方法将任務發送到消息中間件（Broker），Celery Worker 程序監控到該任務後，就會進行執行。我們将視窗切換到 Worker 的啟動視窗，會看到多了兩條日志

INFO/MainProcess] Received task: tasks.add[2272ddce-8be5-493f-b5ff-35a0d9fe600f]

INFO/PoolWorker-4] Task tasks.add[2272ddce-8be5-493f-b5ff-35a0d9fe600f] succeeded in 5.00642602402s: 10

這說明任務已經被排程并執行成功。

另外，我們如果想擷取執行後的結果，可以這樣做：

>>> result = add.delay(, )
>>> result.ready()   # 使用 ready() 判斷任務是否執行完畢
False
>>> result.ready()
False
>>> result.ready()
True
>>> result.get()     # 使用 get() 擷取任務結果

在上面，我們是在 Python 的環境中調用任務。事實上，我們通常在應用程式中調用任務。比如，将下面的代碼儲存為

client.py

from tasks import add

# 異步任務
add.delay(, )

print 'hello world'

運作指令

$ python client.py

，可以看到，雖然任務函數

add

需要等待 5 秒才傳回執行結果，但由于它是一個異步任務，不會阻塞目前的主程式，是以主程式會往下執行

print

語句，列印出結果。

使用配置

在上面的例子中，我們直接把 Broker 和 Backend 的配置寫在了程式當中，更好的做法是将配置項統一寫入到一個配置檔案中，通常我們将該檔案命名為

celeryconfig.py

。Celery 的配置比較多，可以在官方文檔查詢每個配置項的含義。

下面，我們再看一個例子。項目結構如下：

celery_demo                    # 項目根目錄
    ├── celery_app             # 存放 celery 相關檔案
    │   ├── __init__.py
    │   ├── celeryconfig.py    # 配置檔案
    │   ├── task1.py           # 任務檔案 1
    │   └── task2.py           # 任務檔案 2
    └── client.py              # 應用程式

__init__.py

代碼如下：

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

from celery import Celery

app = Celery('demo')                                # 建立 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'                     # 指定時區，預設是 UTC
# CELERY_TIMEZONE='UTC'                             

CELERY_IMPORTS = (                                  # 指定導入的任務子產品
    'celery_app.task1',
    'celery_app.task2'
)

task1.py

代碼如下：

import time
from celery_app import app

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

task2.py

代碼如下：

import time
from celery_app import app

@app.task
def multiply(x, y):
    time.sleep()
    return x * y

client.py

代碼如下：

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

from celery_app import task1
from celery_app import task2

task1.add.apply_async(args=[, ])        # 也可用 task1.add.delay(2, 8)
task2.multiply.apply_async(args=[, ])   # 也可用 task2.multiply.delay(3, 7)

print 'hello world'

現在，讓我們啟動 Celery Worker 程序，在項目的根目錄下執行下面指令：

接着，運作

$ python client.py

，它會發送兩個異步任務到 Broker，在 Worker 的視窗我們可以看到如下輸出：

INFO/MainProcess] Received task: celery_app.task1.add[9ccffad0-aca4-4875-84ce-0ccfce5a83aa]
[INFO/MainProcess] Received task: celery_app.task2.multiply[64b1f889-c892-4333-bd1d-ac667e677a8a]
[INFO/PoolWorker-3] Task celery_app.task1.add[9ccffad0-aca4-4875-84ce-0ccfce5a83aa] succeeded in 2.00600231002s: 10
[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

，如下：

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

支援更多的參數，它的一般形式如下：

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

apply_async 常用的參數如下：

countdown：指定多少秒後執行任務

task1.apply_async(args=(, ), countdown=) #  秒後執行任務

eta (estimated time of arrival)：指定任務被排程的具體時間，參數類型是 datetime

from datetime import datetime, timedelta

# 目前 UTC 時間再加 10 秒後執行任務
task1.multiply.apply_async(args=[, ], eta=datetime.utcnow() + timedelta(seconds=))

expires：任務過期時間，參數類型可以是 int，也可以是 datetime

更多的參數清單可以在官方文檔中檢視。

定時任務

Celery 除了可以執行異步任務，也支援執行周期性任務（Periodic Tasks），或者說定時任務。Celery Beat 程序通過讀取配置檔案的内容，周期性地将定時任務發往任務隊列。

讓我們看看例子，項目結構如下：

celery_demo                    # 項目根目錄
    ├── celery_app             # 存放 celery 相關檔案
        ├── __init__.py
        ├── celeryconfig.py    # 配置檔案
        ├── task1.py           # 任務檔案
        └── task2.py           # 任務檔案

__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'    # 指定時區，不指定預設為 '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 秒執行一次
         'args': (, )                           # 任務函數參數
    },
    'multiply-at-some-time': {
        'task': 'celery_app.task2.multiply',
        'schedule': crontab(hour=, minute=),   # 每天早上 9 點 50 分執行一次
        'args': (, )                            # 任務函數參數
    }
}

task1.py

代碼如下：

import time
from celery_app import app

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

task2.py

代碼如下：

import time
from celery_app import app

@app.task
def multiply(x, y):
    time.sleep()
    return x * y

現在，讓我們啟動 Celery Worker 程序，在項目的根目錄下執行下面指令：

接着，啟動 Celery Beat 程序，定時将任務發送到 Broker，在項目根目錄下執行下面指令：

celery_demo $ celery beat -A celery_app
celery beat v4. (latentcall) is starting.
__    -    ... __   -        _
LocalTime -> -- ::
Configuration ->
    . broker -> redis://127.0.0.1:6379//
    . loader -> celery.loaders.app.AppLoader
    . scheduler -> celery.beat.PersistentScheduler
    . db -> celerybeat-schedule
    . logfile -> [stderr]@%WARNING
    . maxinterval ->  minutes (s)

之後，在 Worker 視窗我們可以看到，任務

task1

每 30 秒執行一次，而

task2

每天早上 9 點 50 分執行一次。

在上面，我們用兩個指令啟動了 Worker 程序和 Beat 程序，我們也可以将它們放在一個指令中：

Celery 周期性任務也有多個配置項，可參考官方文檔。

本文由 funhacks 發表于個人部落格，采用 Creative Commons BY-NC-ND 4.0（自由轉載-保持署名-非商用-禁止演繹）協定釋出。

非商業轉載請注明作者及出處。商業轉載請聯系作者本人。

本文标題為: 異步任務神器 Celery 簡明筆記

本文連結為: https://funhacks.net/2016/12/...

參考資料

Celery - Distributed Task Queue — Celery 4.0.1 documentation
使用Celery - Python之美
分布式任務隊列Celery的介紹 – 思誠之道
異步任務神器 Celery 簡明筆記

Celery 異步分布式 171219 定時任務參考資料

啟動 Celery Worker

調用任務

使用配置

delay 和 apply_async

定時任務

參考資料

繼續閱讀

來自python的【條件控制/語句循環/break/continue/else/pass】一、條件控制二、語句循環

無法解析的外部符号 wmain，該符号在函數 "void cdecl mainCRTStartupHelper(struct HINSTANCE *,unsigned short con......

TestLink導出用例轉換工具(XML2Excel)

YAML簡介和PyYAML安全操作YAML支援的類型YAML的優點：yaml的基本文法python操作

Small tricks

libsvm for python 安裝

學習軟體測試基礎測試第七天

Zeppelin 配置通路 REST APIApache Zeppelin Configuration REST API

【Torch】最簡潔logging使用指南

27. Remove Element(清單)題目代碼

Cloud Studio初體驗

使用 ctypes 進行 Python 和 C 的混合程式設計

【python】【資料處理】畫多元資料分布圖

【python】netconf協定對接管理裝置

「Python 網絡自動化」NETCONF —— Python 使用 NETCONF 管理配置 H3C 網絡裝置

在python中建立excel并寫入

Celery 異步分布式 171219 定時任務 參考資料

啟動 Celery Worker

調用任務

使用配置

delay 和 apply_async

定時任務

參考資料

繼續閱讀

Celery 異步分布式 171219 定時任務參考資料