celery的使用

Celery的定義

Celery（芹菜）是一個簡單、靈活且可靠的，處理大量消息的分布式系統，並且提供維護這樣一個系統的必需工具。

    我比較喜歡的一點是：Celery支持使用任務隊列的方式在分布的機器、進程、線程上執行任務調度。然后我接着去理解什么是任務隊列。

任務隊列

任務隊列是一種在線程或機器間分發任務的機制。

消息隊列

消息隊列的輸入是工作的一個單元，稱為任務，獨立的職程（Worker）進程持續監視隊列中是否有需要處理的新任務。

Celery 用消息通信，通常使用中間人（Broker）在客戶端和職程間斡旋。這個過程從客戶端向隊列添加消息開始，之后中間人把消息派送給職程，職程對消息進行處理。如下圖所示：

 

Celery 系統可包含多個職程和中間人，以此獲得高可用性和橫向擴展能力。

Celery的架構

Celery的架構由三部分組成，消息中間件（message broker），任務執行單元（worker）和任務執行結果存儲（task result store）組成。

消息中間件

Celery本身不提供消息服務，但是可以方便的和第三方提供的消息中間件集成，包括，RabbitMQ,Redis,MongoDB等，這里我先去了解RabbitMQ,Redis。

任務執行單元

Worker是Celery提供的任務執行的單元，worker並發的運行在分布式的系統節點中

任務結果存儲

Task result store用來存儲Worker執行的任務的結果，Celery支持以不同方式存儲任務的結果，包括Redis，MongoDB，Django ORM，AMQP等

celery的簡單示例

首先安裝Celery

pip install celery

因為涉及到消息中間件，所以需要選擇一個broker安裝

RabbitMQ安裝：

參考：http://www.cnblogs.com/cwp-bg/p/8397529.html

redis安裝：

pip install redis

為了測試Celery能否工作，我運行了一個最簡單的任務，編寫tasks.py，如下：

 

import time
from celery import Celery

app = Celery('tasks', broker='redis://192.168.10.48:6379', backend='redis://192.168.10.48:6379')


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

記住：當有多個裝飾器的時候，celery.task一定要在最外層；

編輯保存退出后，我在當前目錄下運行如下命令：

$ celery -A tasks worker --loglevel=info

#查詢文檔，了解到該命令中-A參數表示的是Celery APP的名稱，這個實例中指的就是tasks.py,后面的tasks就是APP的名稱，worker是一個執行任務角色，后面的loglevel=info記錄日志類型默認是info,這個命令啟動了一個worker,用來執行程序中add這個加法任務（task）。

 

發送任務，添加任務到broker

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from s1 import add

# 立即告知celery去執行add任務，並傳入兩個參數
result = add.delay(4, 4)
print(result.id)

備注： 每個任務都會有一個對應該任務的唯一ID，可以通過  result.id 獲取到

 

查看任務執行狀態以及結果

from celery.result import AsyncResult
from s1 import app

async = AsyncResult(id="f0b41e83-99cf-469f-9eff-74c8dd600002", app=app)

if async.successful():
    result = async.get()
    print(result)
    # result.forget() # 將結果刪除
elif async.failed():
    print('執行失敗')
elif async.status == 'PENDING':
    print('任務等待中被執行')
elif async.status == 'RETRY':
    print('任務異常后正在重試')
elif async.status == 'STARTED':
    print('任務已經開始被執行')

 整個celery執行的步驟如下圖：

擴展

• 如果使用redis作為任務隊列中間人，在redis中存在兩個鍵 celery 和 _kombu.binding.celery ， _kombu.binding.celery 表示有一名為 celery 的任務隊列（Celery 默認），而 celery為默認隊列中的任務列表，使用list類型，可以看看添加進去的任務數據。

• 開啟worker

在項目目錄下執行：

celery -A app.celery_tasks.celery worker -Q queue --loglevel=info

• A參數指定celery對象的位置，該app.celery_tasks.celery指的是app包下面的celery_tasks.py模塊的celery實例，注意一定是初始化后的實例，

• Q參數指的是該worker接收指定的隊列的任務，這是為了當多個隊列有不同的任務時可以獨立；如果不設會接收所有的隊列的任務；

• l參數指定worker的日志級別；

執行完畢后結果存儲在redis中，查看redis中的數據，發現存在一個string類型的鍵值對：

celery-task-meta-064e4262-e1ba-4e87-b4a1-52dd1418188f:data

該鍵值對的失效時間為24小時。

分析消息

• 這是添加到任務隊列中的消息數據。

{"body": "gAJ9cQAoWAQAAAB0YXNrcQFYGAAAAHRlc3RfY2VsZXJ5LmFkZF90b2dldGhlcnECWAIAAABpZHEDWCQAAAA2NmQ1YTg2Yi0xZDM5LTRjODgtYmM5OC0yYzE4YjJjOThhMjFxBFgEAAAAYXJnc3EFSwlLKoZxBlgGAAAAa3dhcmdzcQd9cQhYBwAAAHJldHJpZXNxCUsAWAMAAABldGFxCk5YBwAAAGV4cGlyZXNxC05YAwAAAHV0Y3EMiFgJAAAAY2FsbGJhY2tzcQ1OWAgAAABlcnJiYWNrc3EOTlgJAAAAdGltZWxpbWl0cQ9OToZxEFgHAAAAdGFza3NldHERTlgFAAAAY2hvcmRxEk51Lg==",   # body是序列化后使用base64編碼的信息，包括具體的任務參數，其中包括了需要執行的方法、參數和一些任務基本信息
"content-encoding": "binary", # 序列化數據的編碼方式
"content-type": "application/x-python-serialize",  # 任務數據的序列化方式，默認使用python內置的序列化模塊pickle
"headers": {}, 
"properties": 
        {"reply_to": "b7580727-07e5-307b-b1d0-4b731a796652",       # 結果的唯一id
        "correlation_id": "66d5a86b-1d39-4c88-bc98-2c18b2c98a21",  # 任務的唯一id
        "delivery_mode": 2, 
        "delivery_info": {"priority": 0, "exchange": "celery", "routing_key": "celery"},  # 指定交換機名稱，路由鍵，屬性
        "body_encoding": "base64", # body的編碼方式
        "delivery_tag": "bfcfe35d-b65b-4088-bcb5-7a1bb8c9afd9"}}

 

• 將序列化消息反序列化

import pickle
import base64

result = 

base64.b64decode('gAJ9cQAoWAQAAAB0YXNrcQFYGAAAAHRlc3RfY2VsZXJ5LmFkZF90b2dldGhlcnECWAIAAABpZHEDWCQAAAA2NmQ1YTg2Yi0xZDM5LTRjODgtYmM5OC0yYzE4YjJjOThhMjFxBFgEAAAAYXJnc3EFSwlLKoZxBlgGAAAAa3dhcmdzcQd9cQhYBwAAAHJldHJpZXNxCUsAWAMAAABldGFxCk5YBwAAAGV4cGlyZXNxC05YAwAAAHV0Y3EMiFgJAAAAY2FsbGJhY2tzcQ1OWAgAAABlcnJiYWNrc3EOTlgJAAAAdGltZWxpbWl0cQ9OToZxEFgHAAAAdGFza3NldHERTlgFAAAAY2hvcmRxEk51Lg==')
print(pickle.loads(result))

# 結果
{
    'task': 'test_celery.add_together',  # 需要執行的任務
    'id': '66d5a86b-1d39-4c88-bc98-2c18b2c98a21',  # 任務的唯一id
    'args': (9, 42),   # 任務的參數
    'kwargs': {},      
    'retries': 0, 
    'eta': None, 
    'expires': None, # 任務失效時間
    'utc': True, 
    'callbacks': None, # 完成后的回調
    'errbacks': None,  # 任務失敗后的回調
    'timelimit': (None, None), # 超時時間
    'taskset': None, 
    'chord': None
}

 

• 常見的數據序列化方式

binary: 二進制序列化方式；python的pickle默認的序列化方法； json:json 支持多種語言, 可用於跨語言方案，但好像不支持自定義的類對象； XML:類似標簽語言； msgpack:二進制的類 json 序列化方案, 但比 json 的數據結構更小, 更快； yaml:yaml 表達能力更強, 支持的數據類型較 json 多, 但是 python 客戶端的性能不如 json

• 經過比較，為了保持跨語言的兼容性和速度，采用msgpack或json方式；

celery配置

• celery的性能和許多因素有關，比如序列化的方式，連接rabbitmq的方式，多進程、單線程等等；

基本配置項

CELERY_DEFAULT_QUEUE：默認隊列
BROKER_URL  : 代理人的網址
CELERY_RESULT_BACKEND：結果存儲地址
CELERY_TASK_SERIALIZER：任務序列化方式
CELERY_RESULT_SERIALIZER：任務執行結果序列化方式
CELERY_TASK_RESULT_EXPIRES：任務過期時間
CELERY_ACCEPT_CONTENT：指定任務接受的內容序列化類型(序列化)，一個列表；

 

采用配置文件的方式執行celery

# main.py
from celery import Celery
import celeryconfig
app = Celery(__name__, include=["task"])
# 引入配置文件
app.config_from_object(celeryconfig)

if __name__ == '__main__':
    result = add.delay(30, 42)

# task.py
from main import app
@app.task
def add(x, y):
    return x + y  

# celeryconfig.py
BROKER_URL =  'amqp://username:password@localhost:5672/yourvhost'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_TASK_SERIALIZER = 'msgpack'
CELERY_RESULT_SERIALIZER = 'msgpack'
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24   # 任務過期時間
CELERY_ACCEPT_CONTENT = ["msgpack"]            # 指定任務接受的內容類型.

• 一些方法

r.ready()     # 查看任務狀態，返回布爾值, 任務執行完成, 返回 True, 否則返回 False. r.wait() # 等待任務完成, 返回任務執行結果，很少使用； r.get(timeout=1) # 獲取任務執行結果，可以設置等待時間 r.result # 任務執行結果. r.state # PENDING, START, SUCCESS，任務當前的狀態 r.status # PENDING, START, SUCCESS，任務當前的狀態 r.successful # 任務成功返回true r.traceback # 如果任務拋出了一個異常，你也可以獲取原始的回溯信息

celery的裝飾方法celery.task

@celery.task() def name(): pass

• task()方法將任務裝飾成異步，參數：

name:可以顯示指定任務的名字；

serializer：指定序列化的方法；

bind:一個bool值，設置是否綁定一個task的實例，如果把綁定，task實例會作為參數傳遞到任務方法中，可以訪問task實例的所有的屬性，即前面反序列化中那些屬性

@task(bind=True)  # 第一個參數是self，使用self.request訪問相關的屬性 def add(self, x, y): logger.info(self.request.id)

base:定義任務的基類，可以以此來定義回調函數

import celery

class MyTask(celery.Task):
    # 任務失敗時執行
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print('{0!r} failed: {1!r}'.format(task_id, exc))
    # 任務成功時執行
    def on_success(self, retval, task_id, args, kwargs):
        pass
    # 任務重試時執行
    def on_retry(self, exc, task_id, args, kwargs, einfo):
        pass

@task(base=MyTask)
def add(x, y):
    raise KeyError()

exc:失敗時的錯誤的類型；
task_id:任務的id；
args:任務函數的參數；
kwargs:參數；
einfo:失敗時的異常詳細信息；
retval:任務成功執行的返回值；

 

• 另外還可以指定exchange信息等，不過一般不使用；

調用異步任務的方法

task.delay():這是apply_async方法的別名,但接受的參數較為簡單；
task.apply_async(args=[arg1, arg2], kwargs={key:value, key:value})
send_task():可以發送未被注冊的異步任務，即沒有被celery.task裝飾的任務；

# tasks.py
from celery import Celery
app = Celery()
def add(x,y):
    return x+y

app.send_task('tasks.add',args=[3,4])  # 參數基本和apply_async函數一樣
# 但是send_task在發送的時候是不會檢查tasks.add函數是否存在的，即使為空也會發送成功

備注：使用celery執行定時任務時，一定要當前時間轉化為UTC時間，否則就會出問題。

 

• apply_async的參數：

countdown : 設置該任務等待一段時間再執行，單位為s；

eta : 定義任務的開始時間；eta=time.time()+10;

expires : 設置任務時間，任務在過期時間后還沒有執行則被丟棄；

retry : 如果任務失敗后, 是否重試;使用true或false，默認為true

shadow：重新指定任務的名字str，覆蓋其在日志中使用的任務名稱；

retry_policy : 重試策略.

max_retries : 最大重試次數, 默認為 3 次. interval_start : 重試等待的時間間隔秒數, 默認為 0 , 表示直接重試不等待. interval_step : 每次重試讓重試間隔增加的秒數, 可以是數字或浮點數, 默認為 0.2 interval_max : 重試間隔最大的秒數, 即 通過 interval_step 增大到多少秒之后, 就不在增加了, 可以是數字或者浮點數, 默認為 0.2 .

add.apply_async((2, 2), retry=True, retry_policy={ 'max_retries': 3, 'interval_start': 0, 'interval_step': 0.2, 'interval_max': 0.2, })

routing_key:自定義路由鍵；

queue：指定發送到哪個隊列；

exchange：指定發送到哪個交換機；

priority：任務隊列的優先級，0-9之間；

serializer：任務序列化方法；通常不設置；

compression：壓縮方案，通常有zlib, bzip2

headers：為任務添加額外的消息；

link：任務成功執行后的回調方法；是一個signature對象；可以用作關聯任務；

link_error: 任務失敗后的回調方法，是一個signature對象；

• 自定義發布者,交換機,路由鍵, 隊列, 優先級,序列方案和壓縮方法:

task.apply_async((2,2), compression='zlib', serialize='json', queue='priority.high', routing_key='web.add', priority=0, exchange='web_exchange')

一份比較常用的配置文件

# 注意，celery4版本后，CELERY_BROKER_URL改為BROKER_URL
BROKER_URL = 'amqp://username:passwd@host:port/虛擬主機名'
# 指定結果的接受地址
CELERY_RESULT_BACKEND = 'redis://username:passwd@host:port/db'
# 指定任務序列化方式
CELERY_TASK_SERIALIZER = 'msgpack' 
# 指定結果序列化方式
CELERY_RESULT_SERIALIZER = 'msgpack'
# 任務過期時間,celery任務執行結果的超時時間
CELERY_TASK_RESULT_EXPIRES = 60 * 20   
# 指定任務接受的序列化類型.
CELERY_ACCEPT_CONTENT = ["msgpack"]   
# 任務發送完成是否需要確認，這一項對性能有一點影響     
CELERY_ACKS_LATE = True  
# 壓縮方案選擇，可以是zlib, bzip2，默認是發送沒有壓縮的數據
CELERY_MESSAGE_COMPRESSION = 'zlib' 
# 規定完成任務的時間
CELERYD_TASK_TIME_LIMIT = 5  # 在5s內完成任務，否則執行該任務的worker將被殺死，任務移交給父進程
# celery worker的並發數，默認是服務器的內核數目,也是命令行-c參數指定的數目
CELERYD_CONCURRENCY = 4 
# celery worker 每次去rabbitmq預取任務的數量
CELERYD_PREFETCH_MULTIPLIER = 4 
# 每個worker執行了多少任務就會死掉，默認是無限的
CELERYD_MAX_TASKS_PER_CHILD = 40 
# 設置默認的隊列名稱，如果一個消息不符合其他的隊列就會放在默認隊列里面，如果什么都不設置的話，數據都會發送到默認的隊列中
CELERY_DEFAULT_QUEUE = "default" 
# 設置詳細的隊列
CELERY_QUEUES = {
    "default": { # 這是上面指定的默認隊列
        "exchange": "default",
        "exchange_type": "direct",
        "routing_key": "default"
    },
    "topicqueue": { # 這是一個topic隊列 凡是topictest開頭的routing key都會被放到這個隊列
        "routing_key": "topic.#",
        "exchange": "topic_exchange",
        "exchange_type": "topic",
    },
    "task_eeg": { # 設置扇形交換機
        "exchange": "tasks",
        "exchange_type": "fanout",
        "binding_key": "tasks",
    },
    
}

 

-參考：

• http://www.cnblogs.com/cwp-bg/p/8759638.html

• https://www.cnblogs.com/wupeiqi/articles/8796552.html