摘要:最近在研究做消息隊(duì)列時(shí),順便看了一下做消息隊(duì)列的實(shí)現(xiàn)。遠(yuǎn)程連接時(shí)需要認(rèn)證實(shí)例化連接對(duì)象實(shí)例化鏈接參數(shù)對(duì)象創(chuàng)建新的通道模式向綁定到指定的中發(fā)送消息,消費(fèi)者從中取出數(shù)據(jù),類(lèi)似于廣播模式發(fā)布訂閱模式。
最近在研究redis做消息隊(duì)列時(shí),順便看了一下RabbitMQ做消息隊(duì)列的實(shí)現(xiàn)。以下是總結(jié)的RabbitMQ中三種exchange模式的實(shí)現(xiàn),分別是fanout, direct和topic。
base.py:
import pika # 獲取認(rèn)證對(duì)象,參數(shù)是用戶名、密碼。遠(yuǎn)程連接時(shí)需要認(rèn)證 credentials = pika.PlainCredentials("admin", "admin") # BlockingConnection(): 實(shí)例化連接對(duì)象 # ConnectionParameters(): 實(shí)例化鏈接參數(shù)對(duì)象 connection = pika.BlockingConnection(pika.ConnectionParameters( "192.168.0.102", 5672, "/", credentials)) # 創(chuàng)建新的channel(通道) channel = connection.channel()
fanout模式:向綁定到指定exchange的queue中發(fā)送消息,消費(fèi)者從queue中取出數(shù)據(jù),類(lèi)似于廣播模式、發(fā)布訂閱模式。
綁定方式: 在接收端channel.queue_bind(exchange="logs", queue=queue_name)
代碼:
publisher.py:
from base import channel, connection # 聲明exchange, 不聲明queue channel.exchange_declare(exchange="logs", exchange_type="fanout") # 廣播 message = "hello fanout" channel.basic_publish( exchange="logs", routing_key="", body=message ) connection.close()
consumer.py:
from base import channel, connection # 聲明exchange channel.exchange_declare(exchange="logs", exchange_type="fanout") # 不指定queue名字, rabbitmq會(huì)隨機(jī)分配一個(gè)名字, 消息處理完成后queue會(huì)自動(dòng)刪除 result = channel.queue_declare(exclusive=True) # 獲取queue名字 queue_name = result.method.queue # 綁定exchange和queue channel.queue_bind(exchange="logs", queue=queue_name) def callback(ch, method, properties, body): print("body:%s" % body) channel.basic_consume( callback, queue=queue_name ) channel.start_consuming()
direct模式:發(fā)送端綁定一個(gè)routing_key1, queue中綁定若干個(gè)routing_key2, 若key1與key2相等,或者key1在key2中,則消息就會(huì)發(fā)送到這個(gè)queue中,再由相應(yīng)的消費(fèi)者去queue中取數(shù)據(jù)。
publisher.py:
from base import channel, connection channel.exchange_declare(exchange="direct_test", exchange_type="direct") message = "hello" channel.basic_publish( exchange="direct_test", routing_key="info", # 綁定key body=message ) connection.close()
consumer01.py:
from base import channel, connection channel.exchange_declare(exchange="direct_test", exchange_type="direct") result = channel.queue_declare(exclusive=True) queue_name = result.method.queue channel.queue_bind( exchange="direct_test", queue=queue_name, # 綁定的key,與publisher中的相同 routing_key="info" ) def callback(ch, method, properties, body): print("body:%s" % body) channel.basic_consume( callback, queue=queue_name ) channel.start_consuming()
consumer02.py:
from base import channel, connection channel.exchange_declare(exchange="direct_test", exchange_type="direct") result = channel.queue_declare(exclusive=True) queue_name = result.method.queue channel.queue_bind( exchange="direct_test", queue=queue_name, # 綁定的key routing_key="error" ) def callback(ch, method, properties, bosy): print("body:%s" % body) channel.basic_consume( callback, queue=queue_name ) channel.start_consuming()
consumer03.py:
from base import channel, connection channel.exchange_declare(exchange="direct_test", exchange_type="direct") result = channel.queue_declare(exclusive=True) queue_name = result.method.queue key_list = ["info", "warning"] for key in key_list: channel.queue_bind( exchange="direct_test", queue=queue_name, # 一個(gè)queue同時(shí)綁定多個(gè)key,有一個(gè)key滿足條件時(shí)就可以收到數(shù)據(jù) routing_key=key ) def callback(ch, method, properties, body): print("body:%s" % body) channel.basic_consume( callback, queue=queue_name ) channel.start_consuming()
執(zhí)行:
python producer.py python consumer01.py python consumer02.py python consumer03.py
結(jié)果:
consumer01.py: body:b"hello" consumer02.py沒(méi)收到結(jié)果 consumer03.py: body:b"hello"
topic模式不是太好理解,我的理解如下:
對(duì)于發(fā)送端綁定的routing_key1,queue綁定若干個(gè)routing_key2;若routing_key1滿足任意一個(gè)routing_key2,則該消息就會(huì)通過(guò)exchange發(fā)送到這個(gè)queue中,然后由接收端從queue中取出其實(shí)就是direct模式的擴(kuò)展。
綁定方式:
發(fā)送端綁定:
channel.basic_publish( exchange="topic_logs", routing_key=routing_key, body=message )
接收端綁定:
channel.queue_bind( exchange="topic_logs", queue=queue_name, routing_key=binding_key )
publisher.py:
import sys from base import channel, connection # 聲明exchange channel.exchange_declare(exchange="topic_test", exchange_type="topic") # 待發(fā)送消息 message = " ".join(sys.argv[1:]) or "hello topic" # 發(fā)布消息 channel.basic_publish( exchange="topic_test", routing_key="mysql.error", # 綁定的routing_key body=message ) connection.close()
consumer01.py:
from base import channel, connection channel.exchange_declare(exchange="topic_test", exchange_type="topic") result = channel.queue_declare(exclusive=True) queue_name = result.method.queue channel.queue_bind( exchange="topic_test", queue=queue_name, routing_key="*.error" # 綁定的routing_key ) def callback(ch, method, properties, body): print("body:%s" % body) channel.basic_consume( callback, queue=queue_name, no_ack=True ) channel.start_consuming()
consumer02.py:
from base import channel, connection channel.exchange_declare(exchange="topic_test", exchange_type="topic") result = channel.queue_declare(exclusive=True) queue_name = result.method.queue channel.queue_bind( exchange="topic_test", queue=queue_name, routing_key="mysql.*" # 綁定的routing_key ) def callback(ch, method, properties, body): print("body:%s" % body) channel.basic_consume( callback, queue=queue_name, no_ack=True ) channel.start_consuming()
執(zhí)行:
python publisher02.py "this is a topic test" python consumer01.py python consumer02.py
結(jié)果:
consumer01.py的結(jié)果: body:b"this is a topic test" consumer02.py的結(jié)果: body:b"this is a topic test"
說(shuō)明通過(guò)綁定相應(yīng)的routing_key,兩個(gè)消費(fèi)者都收到了消息
將publisher.py的routing_key改成"mysql.info"
再此執(zhí)行:
python publisher02.py "this is a topic test" python consumer01.py python consumer02.py
結(jié)果:
consumer01.py沒(méi)收到結(jié)果 consumer02.py的結(jié)果: body:b"this is a topic test"
通過(guò)這個(gè)例子我們就能明白topic的運(yùn)行方式了。
參考自: https://blog.csdn.net/fgf00/a...
今天就說(shuō)到這里,如有問(wèn)題,歡迎交流指正!
文章版權(quán)歸作者所有,未經(jīng)允許請(qǐng)勿轉(zhuǎn)載,若此文章存在違規(guī)行為,您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請(qǐng)注明本文地址:http://specialneedsforspecialkids.com/yun/44948.html
摘要:需要特別明確的概念交換機(jī)的持久化,并不等于消息的持久化。消息的處理,是有兩種方式,一次性。在上述示例中,使用的,意味著接收全部的消息。注意與是兩個(gè)不同的隊(duì)列。后端處理,可以針對(duì)每一個(gè)啟動(dòng)一個(gè)或多個(gè),以提高消息處理的實(shí)時(shí)性。 RabbitMQ與PHP(一) 項(xiàng)目中使用RabbitMQ作為隊(duì)列處理用戶消息通知,消息由前端PHP代碼產(chǎn)生,處理消息使用Python,這就導(dǎo)致代碼一致性問(wèn)題,調(diào)...
摘要:在發(fā)送后端監(jiān)聽(tīng)聲明的排他隊(duì)列,當(dāng)收到消息后比對(duì)正確則處理消息斷開(kāi)監(jiān)聽(tīng)連接,然后此隊(duì)列被系統(tǒng)自動(dòng)回收。并且通過(guò)也看到了這條消息的返回。此時(shí)我們基本已經(jīng)將問(wèn)題鎖定在端了。 背景 公司的一個(gè)項(xiàng)目使用rabbitmq作為broker進(jìn)行交互,并且數(shù)據(jù)的查詢方法使用RPC模式,RPC Client端使用java編寫(xiě)并使用springAMQP包與rabbitmq交互,在RPC Server端使用p...
摘要:消息隊(duì)列選擇是一個(gè)由開(kāi)發(fā)的的開(kāi)源實(shí)現(xiàn)的產(chǎn)品,是一個(gè)消息代理,從生產(chǎn)者接收消息并傳遞消息至消費(fèi)者,期間可根據(jù)規(guī)則路由緩存持久化消息。綁定隊(duì)列和交換機(jī)之間的關(guān)系。根據(jù)消息的屬性和的屬性來(lái)轉(zhuǎn)發(fā)消息。 消息隊(duì)列選擇:RabbitMQ & Redis RabbitMQ RabbitMQ是一個(gè)由erlang開(kāi)發(fā)的AMQP(Advanced Message Queue )的開(kāi)源實(shí)現(xiàn)的產(chǎn)品,Rabbi...
摘要:每當(dāng)我們收到一條消息,這個(gè)回調(diào)函數(shù)就被皮卡庫(kù)調(diào)用。接下來(lái),我們需要告訴這個(gè)特定的回調(diào)函數(shù)應(yīng)該從我們的隊(duì)列接收消息為了讓這個(gè)命令成功,我們必須確保我們想要訂閱的隊(duì)列存在。生產(chǎn)者計(jì)劃將在每次運(yùn)行后停止歡呼我們能夠通過(guò)發(fā)送我們的第一條消息。 源碼:https://github.com/ltoddy/rabbitmq-tutorial 介紹 RabbitMQ是一個(gè)消息代理:它接受和轉(zhuǎn)發(fā)消息。你...
閱讀 3648·2021-10-09 09:58
閱讀 1188·2021-09-22 15:20
閱讀 2495·2019-08-30 15:54
閱讀 3510·2019-08-30 14:08
閱讀 887·2019-08-30 13:06
閱讀 1817·2019-08-26 12:16
閱讀 2678·2019-08-26 12:11
閱讀 2507·2019-08-26 10:38