django+celery+RabbitMQ自定義多個消息隊列的實現(xiàn)
更新時間：2023年02月22日 15:53:55 作者：Hello_Mr_Zheng
本文主要介紹了django+celery+RabbitMQ自定義多個消息隊列的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
關于django celery的使用網(wǎng)上有很多文章，本文就不多做更多的說明。
本文使用版本
- python==3.8.15
- Django==3.2.4
- celery==5.2.7
celery.py
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from kombu import Exchange, Queue
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'zkcelery.settings')
app = Celery('zkcelery')
# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
# should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')
# Load task modules from all registered Django app configs.
app.autodiscover_tasks()
# 看了一篇文章說，如果使用redis做broker，exchange可以不配置；但如果使用rabbitMQ做broker，就必須要配置。
queue = (
Queue('default', exchange=Exchange('default', type='direct'), routing_key='default'),
Queue('q1', exchange=Exchange('e1', type='direct'), routing_key='r1'),
Queue('q2', exchange=Exchange('e2', type='direct'), routing_key='r2'),
Queue('q3', exchange=Exchange('e3', type='fanout'), routing_key='r3'),
)
# 一旦配置了route后，所有的任務名都必須要指定route，否則任務無法執(zhí)行。
# 經過測試，route匹配是最長匹配規(guī)則。
route = {
'apps.zhiding.tasks.add': {'queue': 'q1', 'routing_key': 'r1'},
'apps.zhiding.tasks.multiply': {'queue': 'q2', 'routing_key': 'r2'},
# 其它的任務名稱，匹配這條路由
# 如果以上隊列的worker服務器壞了，這些任務會被全部放進這個隊列里，該隊列的worker將繼續(xù)處理這些任務
# 下面這條隊列一定要配置，否則其它任務無法處理。
'*': {'queue': 'default', 'routing_key': 'default'},
}
app.conf.update(CELERY_QUEUES=queue, CELERY_ROUTES=route)
tasks.py
from celery import shared_task
import time
@shared_task
def add(x, y):
time.sleep(2)
print('任務睡眠2秒后執(zhí)行了')
return x + y
@shared_task
def multiply(x, y):
time.sleep(5)
print('任務睡眠5秒后執(zhí)行了')
return x * y
@shared_task
def sub(x, y):
time.sleep(4)
print('任務睡眠4秒后執(zhí)行了')
return x - y
筆者也看了很多博文，在settings.py配置文件中寫入CELERY_QUEUES和CELERY_ROUTES，上面的配置對應下來就是如下代碼塊：
CELERY_QUEUES = (
Queue('default', exchange=Exchange('default', type='direct'), routing_key='default'),
Queue('sq1', exchange=Exchange('sq1', type='direct'), routing_key='sq1'),
Queue('sq2', exchange=Exchange('sq2', type='direct'), routing_key='sq2'),
Queue('sq3', exchange=Exchange('sq3', type='fanout'), routing_key='sq3'),
)
CELERY_ROUTES = {
'apps.zhiding.tasks.add': {'queue': 'sq1', 'routing_key': 'sq1'},
'apps.zhiding.tasks.multiply': {'queue': 'sq2', 'routing_key': 'sq2'},
'*': {'queue': 'default', 'routing_key': 'default'},
}
但是筆者在實際使用中發(fā)現(xiàn)后面這種方式配置始終未生效，不知道是不是筆者版本的不同，沒有做更多的研究，如果你能找到問題的原因，歡迎評論交流。
啟動worker
# 筆者使用的windows，啟動時需要加上-P eventlet celery -A zkcelery worker -l info -P eventlet
啟動后隊列中出現(xiàn)配置中的個隊列
同時會在rabbitmq中創(chuàng)建（如果不存在）4個隊列，交換機和相應的綁定關系（當然也可以直接通過rabbitmq管理端直接創(chuàng)建自己需要的隊列、交換機和綁定，具體根據(jù)個人習慣或者視工作場景而定選擇）
以隊列q1示例：
暫時先關閉worker，便于觀察消息隊列中的消息。
向隊列中發(fā)送幾條消息，消息均進入到配置中指定的queue中
再次啟動worker，隊列中的消息立馬被消費
如何做到消費指定的隊列中的消息，只需要啟動的時候加上參數(shù)Q
# -Q指定消費的隊列 # -n 指定worker節(jié)點的名稱，避免啟動多個時的重名沖突 celery -A zkcelery worker -l info -Q q1 -n node1 -P eventlet
可以看到終端中queues只有q1了
q1中的消息被消費掉了，其他隊列沒有變化
也可以同時指定多個消費隊列
celery -A zkcelery worker -l info -Q q2,default -n node2 -P eventlet
當然也可以在生產方指定推送的隊列，舉例如下：
到此這篇關于django+celery+RabbitMQ自定義多個消息隊列的實現(xiàn)的文章就介紹到這了,更多相關django celery RabbitMQ消息隊列內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
這篇文章主要介紹了解決python3安裝pandas出錯的問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-05-05
Python3導入CSV文件的實例(跟Python2有些許的不同)
今天小編就為大家分享一篇Python3導入CSV文件的實例(跟Python2有些許的不同)，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-06-06
anaconda中Conda創(chuàng)建虛擬環(huán)境的實現(xiàn)步驟
在Anaconda中,可以使用conda命令來創(chuàng)建和管理虛擬環(huán)境,本文主要介紹了anaconda中Conda創(chuàng)建虛擬環(huán)境的實現(xiàn)步驟,具有一定的參考價值,感興趣的可以了解一下2023-12-12
這篇文章主要介紹了Python基于類路徑字符串獲取靜態(tài)屬性,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-03-03
這篇文章主要介紹了python文件讀取時順序錯誤的問題及解決方案,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-08-08
python tkinter圖形界面代碼統(tǒng)計工具
這篇文章主要為大家詳細介紹了python tkinter圖形界面代碼統(tǒng)計工具，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-09-09
python3+telnetlib實現(xiàn)簡單自動測試示例詳解
telnetlib 模塊提供一個實現(xiàn)Telnet協(xié)議的類 Telnet，本文重點給大家介紹python3+telnetlib實現(xiàn)簡單自動測試示例詳解,需要的朋友可以參考下2021-08-08
這篇文章主要介紹了django創(chuàng)建超級用戶過程解析,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2019-09-09
Matplotlib是Python中最受歡迎的數(shù)據(jù)可視化軟件包之一，它是?Python常用的2D繪圖庫，同時它也提供了一部分3D繪圖接口。本文將詳細介紹Matplotlib的繪圖方式，需要的可以參考一下2022-07-07
基于Python繪制美觀動態(tài)圓環(huán)圖、餅圖
這篇文章主要介紹了基于Python制作美觀動態(tài)圓環(huán)圖、餅圖,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-06-06