異步任務神器 Celery 簡明筆記

Ryan_Li 發布于2019-07-25 11:13 / 1565人閱讀

摘要：我們將窗口切換到的啟動窗口，會看到多了兩條日志這說明任務已經被調度并執行成功。本文標題為異步任務神器簡明筆記本文鏈接為參考資料使用之美分布式任務隊列的介紹思誠之道異步任務神器簡明筆記

Celery

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

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

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

任務模塊
包含異步任務和定時任務。其中，異步任務通常在業務邏輯中被觸發并發往任務隊列，而定時任務由 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 及相關依賴：

$ pip install "celery[redis]"

創建 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(5)     # 模擬耗時操作
    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：

$ celery worker -A tasks --loglevel=info

其中：

參數 -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(2, 8)

在上面，我們從 tasks.py 文件中導入了 add 任務對象，然后使用 delay() 方法將任務發送到消息中間件（Broker），Celery Worker 進程監控到該任務后，就會進行執行。我們將窗口切換到 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

這說明任務已經被調度并執行成功。

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

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

在上面，我們是在 Python 的環境中調用任務。事實上，我們通常在應用程序中調用任務。比如，將下面的代碼保存為 client.py:

# -*- 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 的配置比較多，可以在官方文檔查詢每個配置項的含義。

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

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(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"

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

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

接著，運行 $ python client.py，它會發送兩個異步任務到 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

在前面的例子中，我們使用 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=(2, 3), countdown=5)    # 5 秒后執行任務

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

from datetime import datetime, timedelta

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

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

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

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

定時任務

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),       # 每 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 代碼如下：

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

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

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

接著，啟動 Celery Beat 進程，定時將任務發送到 Broker，在項目根目錄下執行下面命令：

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 進程，我們也可以將它們放在一個命令中：

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

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 簡明筆記

云服務器 GPU云服務器 Django下使用celery 異步發送異步任務隊列 asynctask異步任務 celery

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/38277.html

分布式隊列神器 Celery

摘要：是什么是一個由編寫的簡單靈活可靠的用來處理大量信息的分布式系統它同時提供操作和維護分布式系統所需的工具。專注于實時任務處理，支持任務調度。說白了，它是一個分布式隊列的管理工具，我們可以用提供的接口快速實現并管理一個分布式的任務隊列。 Celery 是什么? Celery 是一個由 Python 編寫的簡單、靈活、可靠的用來處理大量信息的分布式系統,它同時提供操作和維護分布式系統所需的工...

趙春朋 2019-07-25 11:19 評論0 收藏0
Python之celery的簡介與使用

摘要：的簡介是一個基于分布式消息傳輸的異步任務隊列，它專注于實時處理，同時也支持任務調度。目前支持等作為消息代理，但適用于生產環境的只有和官方推薦。任務處理完后保存狀態信息和結果，以供查詢。 celery的簡介 ??celery是一個基于分布式消息傳輸的異步任務隊列，它專注于實時處理，同時也支持任務調度。它的執行單元為任務（task），利用多線程，如Eventlet，gevent等，它們能被...

LeexMuller 2019-07-31 10:02 評論0 收藏0
代寫文章列表

摘要：技術配置與美化方案開發環境搭建自定義腳手架關系型數據庫簡明用法緩存系統使用部署開發異步任務系統進行文檔編寫進行代碼測試在中加入支付功能開源搜索引擎簡明用法開源日志引擎簡明用法開源監控系統簡明用法網絡爬蟲編寫簡明指南優雅進行代碼管理思路接口文技術 Ubuntu配置與美化方案 Python開發環境搭建自定義Django腳手架關系型數據庫簡明用法 Django緩存系統使用 Fabri...

GitCafe 2019-07-25 11:30 評論0 收藏0
基于Flask-Angular的項目組網架構與部署

摘要：基于網，分享項目的組網架構和部署。項目組網架構架構說明流項目訪問分為兩個流，通過分兩個端口暴露給外部使用數據流用戶訪問網站。通過進行配置，使用作為異步隊列來存儲任務，并將處理結果存儲在中。基于Raindrop網，分享項目的組網架構和部署。項目組網架構 showImg(https://cloud.githubusercontent.com/assets/7239657/1015704...

kelvinlee 2019-06-24 17:48 評論0 收藏0
Django下使用celery 異步發送短信驗證碼

摘要：介紹應用舉例是一個基于開發的分布式異步消息任務隊列，通過它可以輕松的實現任務的異步處理，如果你的業務場景中需要用到異步任務，就可以考慮使用你想對臺機器執行一條批量命令，可能會花很長時間，但你不想讓你的程序等著結果返回，? celery 1.celery介紹 1.1 celery應用舉例 Celery 是一個基于python開發的分布式異步消息任務隊列，通過...

everfly 2021-09-02 15:12 評論0 收藏0