2. 程式人生 > >python之celery使用詳解一

python之celery使用詳解一

阿新 • • 發佈：2018-11-13

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

celery + rabbitmq初步

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

簡單介紹

Celery 是一個非同步任務佇列。一個Celery安裝有三個核心元件：

  1. Celery 客戶端: 用於釋出後臺作業。當與 Flask 一起工作的時候，客戶端與 Flask 應用一起執行。

  2. Celery workers: 執行後臺作業的程序。Celery 支援本地和遠端的 workers，可以在 Flask 伺服器上啟動一個單獨的 worker，也可以在遠端伺服器上啟動worker，需要拷貝程式碼；

  3. 訊息代理: 客戶端通過訊息佇列和 workers 進行通訊，Celery 支援多種方式來實現這些佇列。最常用的代理就是 RabbitMQ 和 Redis。

安裝rabbitmq和redis

sudo pip install redis
sudo pip install celery[librabbitmq]

初步使用

  • 使用redis做結果儲存，使用rabbitmq做任務佇列；
# tasks.py
from celery import Celery

app = Celery('tasks', broker='amqp://username:
 
[email protected]:port/varhost',backend='redis://username:[email protected]: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:[email protected]: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:[email protected]:port/虛擬主機名'
# 指定結果的接受地址
CELERY_RESULT_BACKEND = 'redis://username:[email protected]: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.#",
    }
}

 

-參考：

閱讀原文

相關推薦

Python之Metaclass詳解，Python之元類

turned 除了 方法 寫法 找到 類對象 global 所在 code 本人Java程序員一枚，這幾天閑來無事就自學了下Python，學到Metaclass感覺有點迷惑，就在網上查相關資料，在棧溢出(stackoverflow)網站上看到一個關於metaclass的回答

Python之pexpect詳解

控制 con 是否 差距 自動化 pytho out 匹配 id_rsa 一、引子 Pexpect程序主要用於人機對話的模擬，就是那種系統提問，人來回答yes/no，或者賬號登陸輸入用戶名和密碼等等的情況。因為這種情況特別多而且繁瑣，所以很多語言都有各種自己的實現。最初的第

Java8新特性之Stream詳解一

寫了Lambda表示式的部落格，怎麼能少了對Stream的學習呢！接下來就將我所理解的Stream分享給大家，歡迎指出不足之處... 一、為什麼要使用Stream？Stream是什麼？在Java中該怎麼用呢？        (1)為什麼要使用Str

Python之序列詳解

序列 一種資料結構，成員有序排列，可通過下標偏移量訪問它的一或多個成員。 序列型別 字串（普通字串和unicode字串)、列表和元組 序列型別操作符 成員關係操作符（in，not in) 作用：判斷一個元素是否屬於一個序列        語法格式：物件

Python之TCP詳解和 OSI七層模型

1.OSI七層模型和TCP/IP四層 基本模型： OSI七層模型 先有模型，後有協議，先有標準，後有實踐，TCP/IP反之 ARP協議，獲取主機的mac地址，全世界唯一 應用程式：QQ、微信，我們開發

RxJava2總結之操作詳解(一)

目錄 建立操作 create 建立一個具有發射能力的Observable Observable.create(e -> { e.onNext("Love"); e.onNext("For"); e.onNe

Android開發學習之SQLite詳解一

一.SQLite的介紹 1.SQLite簡介  SQLite是一款輕型的資料庫，是遵守ACID的關聯式資料庫管理系統，它的設計目標是嵌入 式的，而且目前已經在很多嵌入式產品中使用了它，它佔用資源非常的低，在嵌入式裝置中，可能只需要幾百K的記憶體就夠了。它能夠支援 Windo

04 python之函式詳解

一、函式初識 函式的產生：函式就是封裝一個功能的程式碼片段。 li = ['spring', 'summer', 'autumn', 'winter'] def function(): count = 0 for j in li: count += 1 print(c

python之celery使用詳解一

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

跟我學python之mysql操作詳解 一、查詢操作

  在python中，我們使用PyMySQL來操作mysql。   一、安裝PyMySQL：    使用pip進行安裝，命令為：    pip install PyMySQL   二、在進行操作之前，首先要開啟本機的mysql，或者，可以直接連線遠端已經開啟的mysql。

python模塊之XlsxWriter 詳解

som hnu ima conf all 合並單元格 stack from pen Xlsx是python用來構造xlsx文件的模塊，可以向excel2007+中寫text，numbers，formulas 公式以及hyperlinks超鏈接。 可以完成xlsx文件的自動化

大數據入門第十七天——storm上遊數據源 之kafka詳解（一）入門

不同 這也 接受 blog 存儲 發送 records ant post 一、概述 　　1.kafka是什麽 　　　　根據標題可以有個概念：kafka是storm的上遊數據源之一，也是一對經典的組合，就像郭德綱和於謙 　　　　根據官網：http://kafka.apa

三：python 對象類型詳解一：數字（上）

結果 dom 運行 精度 升級 方法 函數 般的 代碼 一：python 的數字類型： a)整數和浮點數 b)復數 c)固定精度的十進制數 d)有理分數 e)集合 f)布爾類型 g)無窮的整數精度 h)各種數字內置函數和模塊 二：各種數字類型的詳解 　　1，數字常量：pyt

JVM之JMM詳解（卷一）

1 序言：    在講解Java的記憶體模型之前，我們可以看看物理機在遇到併發問題時所想出的處理方案（物理機和Java虛擬機器在解決併發問題有不少相似的地方，具有很大的參考價值）。本章旨在剖析Java的記憶體模型以及讓讀者明白Java併發程式設計裡不知其所以然的地方。（推薦

python模組 - psutil python模組之psutil詳解

一、psutil模組: 1. psutil模組簡介         他是一個跨平臺庫 ( http://pythonhosted.org/psutil/ ) 能夠輕鬆實現獲取系統執行的程序和系統利用率（包括CPU、記憶體、磁碟、網路等）資訊。它主

Python yield使用詳解（一）

生成器 yield語句可以作為生成器 def countdown(n): while n > 0: yield n n -= 1 # 可以當迭代器來使用它 for x in countdown(10): print('T

一看就懂，Python 日誌模組詳解及應用

日誌概述 百度百科的日誌概述： Windows網路作業系統都設計有各種各樣的日誌檔案，如應用程式日誌，安全日誌、系統日誌、Scheduler服務日誌、FTP日誌、WWW日誌、DNS伺服器日誌等等，這些根據你的系統開啟的服務的不同而有所不同。我們在系統上進行一些操作時，這些日誌檔案通常會記錄下我們操作的一些

docker系列詳解<一>之docker安裝

1、Docker 要求 CentOS 系統的核心版本高於 3.10 ，檢視本頁面的前提條件來驗證你的CentOS 版本是否支援 Docker 。 通過 uname -r 命令檢視你當前的核心版本 $ uname -r 2、使用 root 許可權登入 Centos。確保 yum 包更新到最新。 $ su

死磕Netty原始碼之記憶體分配詳解(一)(PooledByteBufAllocator)

前言 為了避免頻繁的記憶體分配給系統帶來負擔以及GC對系統性能帶來波動，Netty4使用了記憶體池來管理記憶體的分配和回收，Netty記憶體池參考了Slab分配和Buddy分配思想。Slab分配是將記憶體分割成大小不等的記憶體塊，在使用者執行緒請求時根據請求的

死磕Netty原始碼之Reactor執行緒模型詳解(一)NioEventLoop的啟動

前言 Netty中最核心的就是Reactor執行緒，對應Netty中的程式碼就是NioEventLoop。NioEventLoop是通過NioEventLoopGroup進行維護的，所以在介紹NioEventLoop前我們先介紹一下NioEventLoopGr