python之celery使用詳解一

本文轉載自查看原文 2018-11-13 19:10 1824 python

前段時間需要使用rabbitmq做寫緩存，一直使用pika+rabbitmq的組合，pika這個模塊雖然可以很直觀地操作rabbitmq，但是官方給的例子太簡單，對其底層原理了解又不是很深，遇到很多坑，尤其是需要自己寫連接池管理和channel池管理。雖然也有用過celery，一直也是celery+redis的組合，涉及很淺；目前打算深研一下celery+redis+rabbitmq的使用。

celery + rabbitmq初步

我們先不在集成框架如flask或Django中使用，而僅僅單獨使用。

簡單介紹

Celery 是一個異步任務隊列。一個Celery安裝有三個核心組件：

Celery 客戶端: 用於發布后台作業。當與 Flask 一起工作的時候，客戶端與 Flask 應用一起運行。
Celery workers: 運行后台作業的進程。Celery 支持本地和遠程的 workers，可以在 Flask 服務器上啟動一個單獨的 worker，也可以在遠程服務器上啟動worker，需要拷貝代碼；
消息代理: 客戶端通過消息隊列和 workers 進行通信，Celery 支持多種方式來實現這些隊列。最常用的代理就是 RabbitMQ 和 Redis。

安裝rabbitmq和redis

rabbitmq安裝和配置參考：http://www.cnblogs.com/cwp-bg/p/8397529.html
redis的安裝和配置參考：http://www.cnblogs.com/cwp-bg/p/8094914.html
redis-py安裝：

sudo pip install redis

redis-py操作redis參考：http://www.cnblogs.com/cwp-bg/p/8274269.html
為了提高性能，官方推薦使用librabbitmq，這是一個連接rabbitmq的C++的庫；

sudo pip install celery[librabbitmq]

初步使用

使用redis做結果存儲，使用rabbitmq做任務隊列；

# tasks.py from celery import Celery app = Celery('tasks', broker='amqp://username:passwd@ip:port/varhost',backend='redis://username:passwd@ip:6390/db') @app.task def add(x, y): return x + y if __name__ == '__main__': result = add.delay(30, 42)

broker:任務隊列的中間人；
backend:任務執行結果的存儲；

發生了什么事

app.task裝飾后將add函數變成一個異步的任務，add.delay函數將任務序列化發送到rabbitmq；
該過程創建一個名字為celery的exchange，類型為direct（直連交換機）;創建一個名為celery的queue，隊列和交換機使用路由鍵celery綁定；
打開rabbitmq管理后台，可以看到有一條消息已經在celery隊列中；

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

擴展

如果使用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函數是否存在的，即使為空也會發送成功

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", }, }
# 不同task指定queue
CELERY_ROUTES = {
    "projq.tasks.add": { # task函數名，必須是全路徑 app.tasks.method_name
        "queue": "topicqueue", 
        "routing_key": "topic.#",
    }
}

-參考：

http://docs.celeryproject.org/en/latest/userguide/tasks.html#task-options
http://docs.jinkan.org/docs/flask/patterns/celery.html
http://www.pythondoc.com/flask-celery/first.html
https://blog.csdn.net/kk123a/article/details/74549117
https://blog.csdn.net/preyta/article/details/54288870
Celery消息隊列----路由任務（含task指定queue）
分布式任務隊列 Celery 的使用（含task指定queue）

閱讀原文

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 python之celery使用詳解一 python之celery使用詳解(二) python之celery使用詳解一 celery 使用詳解 Python—Celery 框架使用 Python的Celery使用 python中的Celery基本使用 python celery介紹和基本使用 Python之celery的簡介與使用 Python—在Django中使用Celery