快捷導(dǎo)航

Python Celery多隊(duì)列配置代碼實(shí)例

更新時(shí)間：2019年11月22日 09:07:09 作者：-零

這篇文章主要介紹了Python Celery多隊(duì)列配置代碼實(shí)例,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

Celery官方文檔

項(xiàng)目結(jié)構(gòu)

/proj
-__init__
-app.py            #實(shí)例化celery對(duì)象
-celeryconfig.py        #celery的配置文件
-tasks.py           #celery編寫任務(wù)文件

app.py

#coding:utf-8
from __future__ import absolute_import
from celery import Celery

app = Celery('proj', include=['proj.tasks'])   #實(shí)例化celery對(duì)象

app.config_from_object('proj.celeryconfig')   #引入配置文件

if __name__ == '__main__':            
  app.start()

proj參數(shù)為celery的名字
include參數(shù)為啟動(dòng)時(shí)導(dǎo)入的模塊列表

tasks.py

#coding:utf-8
from __future__ import absolute_import

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

celeryconfig.py

#coding:utf-8
from kombu import Queue

BROKER_URL = 'amqp://guest:guest@127.0.0.1:5672//' # 使用RabbitMQ作為消息代理


CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0' # 把任務(wù)結(jié)果存在了Redis

CELERY_TASK_SERIALIZER = 'msgpack' # 任務(wù)序列化和反序列化使用msgpack方案

CELERY_RESULT_SERIALIZER = 'json' # 讀取任務(wù)結(jié)果一般性能要求不高，所以使用了可讀性更好的JSON

CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # 任務(wù)過期時(shí)間，不建議直接寫86400，應(yīng)該讓這樣的magic數(shù)字表述更明顯

CELERY_ACCEPT_CONTENT = ['json', 'msgpack'] # 指定接受的內(nèi)容類型

CELERY_QUEUES = (  #設(shè)置add隊(duì)列,綁定routing_key
  Queue('add', routing_key='xue.add'),
)


CELERY_ROUTES = {  #projq.tasks.add這個(gè)任務(wù)進(jìn)去add隊(duì)列并routeing_key為xue.add 
  'projq.tasks.add': { 
    'queue': 'add',
    'routing_key': 'xue.add',
  }
}

CELERY_ACCEPT_CONTENT的類型msgpack為是一種比json更小更快的類型，如果用需要安裝相對(duì)應(yīng)的包。
CELERY_QUEUES設(shè)置一個(gè)指定routing_key的隊(duì)列，這個(gè)名字可以任意指定。
CELERY_ROUTES設(shè)置路由，對(duì)指定的任務(wù)名，指定對(duì)應(yīng)的隊(duì)列和routing_key，注意，這里的routing_key需要和上面參數(shù)的一致。

啟動(dòng)

在proj的上層目錄輸入

celery -A proj.app worker -Q add -l info

proj.tasks.add為任務(wù)名稱，也就是在CELERY_ROUTES設(shè)置的那個(gè)名稱

add是設(shè)置的queue，key=xue.add是設(shè)置的routing_key

發(fā)布任務(wù)

from proj.tasks import add
add.delay(2,3)

多隊(duì)列中需要修改的地方

CELERY_QUEUES = (  #設(shè)置add隊(duì)列,綁定routing_key
  Queue('add', routing_key='xue.add'),
)


CELERY_ROUTES = {  #projq.tasks.add這個(gè)任務(wù)進(jìn)去add隊(duì)列并routeing_key為xue.add 
  'projq.tasks.add': { 
    'queue': 'add',
    'routing_key': 'xue.add',
  }

配置兩個(gè)隊(duì)列

# 配置隊(duì)列
CELERY_QUEUES = (
  Queue('default', routing_key='default'),
  Queue('隊(duì)列1', routing_key='key1'),
  Queue('隊(duì)列2', routing_key='key2'),
)

# 路由（哪個(gè)任務(wù)放入哪個(gè)隊(duì)列）
CELERY_ROUTES = {
  '任務(wù)1': {'queue': '隊(duì)列1', 'routing_key': 'key1'},
  '任務(wù)2': {'queue': '對(duì)列2', 'routing_key': 'key2'},
}

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: