異步任務神器 Celery－入門

時間  2019-11-06
標籤 異步 任務 神器 celery 入門 简体版
原文   原文鏈接

1、Celery入門介紹

　　在程序的運行過程當中，咱們常常會碰到一些耗時耗資源的操做，爲了不它們阻塞主程序的運行，咱們常常會採用多線程或異步任務。好比，在 Web 開發中，對新用戶的註冊，咱們一般會給他發一封激活郵件，而發郵件是個 IO 阻塞式任務，若是直接把它放到應用當中，就須要等郵件發出去以後才能進行下一步操做，此時用戶只能等待再等待。更好的方式是在業務邏輯中觸發一個發郵件的異步任務，而主程序能夠繼續往下運行。html

Celery 是一個強大的分佈式任務隊列，它可讓任務的執行徹底脫離主程序，甚至能夠被分配到其餘主機上運行。咱們一般使用它來實現異步任務（async task）和定時任務（crontab）。它的架構組成以下圖：python

 

能夠看到，Celery 主要包含如下幾個模塊：git

  • 任務模塊 Taskgithub

    包含異步任務和定時任務。其中，異步任務一般在業務邏輯中被觸發併發往任務隊列，而定時任務由 Celery Beat 進程週期性地將任務發往任務隊列。redis

  • 消息中間件 Broker多線程

    Broker，即爲任務調度隊列，接收任務生產者發來的消息（即任務），將任務存入隊列。Celery 自己不提供隊列服務，官方推薦使用 RabbitMQ 和 Redis 等。架構

  • 任務執行單元 Worker併發

    Worker 是執行任務的處理單元，它實時監控消息隊列，獲取隊列中調度的任務，並執行它。app

  • 任務結果存儲 Backend異步

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

 

2、異步任務Celery快速入門

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

  1. 建立一個 Celery 實例
  2. 啓動 Celery Worker
  3. 應用程序調用異步任務

 

一、安裝Celery

爲了簡單起見，對於 Broker 和 Backend，這裏都使用 redis。在運行下面的例子以前，請確保 redis 已正確安裝，並開啓 redis 服務，固然，celery 也是要安裝的。可使用下面的命令來安裝 celery 及相關依賴：

$ 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 調度的任務；

 

三、啓動 Celery Worker

在當前目錄，使用以下方式啓動 Celery Worker：

1
$ celery worker -A tasks.app --loglevel=info

其中：

  • 參數 -A 指定了 Celery 實例的位置，本例是在 tasks.py 中，Celery 會自動在該文件中尋找 Celery 對象實例，固然，咱們也能夠本身指定，在本例，使用 -A tasks.app；
  • 參數 --loglevel 指定了日誌級別，默認爲 warning，也可使用 -l info 來表示；

在生產環境中，咱們一般會使用 Supervisor 來控制 Celery Worker 進程，這篇博客，不介紹Supervisor，其實也簡單，是一個控制進程的工具，還有可視化界面，以後會寫一篇博客來介紹Celery實戰，並用上Supervisor。

啓動成功後，控制檯會顯示以下輸出：

 

四、調用任務

如今，咱們能夠在應用程序中使用 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 秒才返回執行結果，但因爲它是一個異步任務，不會阻塞當前的主程序，所以主程序並不會等待5秒再往下執行 print 語句，打印出結果。

 

3、使用配置

在上面的例子中，咱們直接把 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 秒後過時

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

 

4、定時任務

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 週期性任務也有多個配置項，可參考官方文檔。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程