最近在研究redis做消息隊列時，順便看了一下RabbitMQ做消息隊列的實現(xiàn)。以下是總結(jié)的RabbitMQ中三種exchange模式的實現(xiàn)，分別是fanout, direct和topic。
base.py：
import pika# 獲取認(rèn)證對象，參數(shù)是用戶名、密碼。遠(yuǎn)程連接時需要認(rèn)證credentials = pika.PlainCredentials("admin", "admin")# BlockingConnection(): 實例化連接對象# ConnectionParameters(): 實例化鏈接參數(shù)對象connection = pika.BlockingConnection(pika.ConnectionParameters( "192.168.0.102", 5672, "/", credentials))# 創(chuàng)建新的channel(通道)channel = connection.channel() fanout模式：向綁定到指定exchange的queue中發(fā)送消息，消費者從queue中取出數(shù)據(jù)，類似于廣播模式、發(fā)布訂閱模式。
綁定方式: 在接收端channel.queue_bind(exchange="logs", queue=queue_name)
代碼：
publisher.py:
from base import channel, connection# 聲明exchange, 不聲明queuechannel.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 # 聲明exchangechannel.exchange_declare(exchange="logs", exchange_type="fanout")# 不指定queue名字, rabbitmq會隨機(jī)分配一個名字, 消息處理完成后queue會自動刪除result = channel.queue_declare(exclusive=True) # 獲取queue名字queue_name = result.method.queue# 綁定exchange和queuechannel.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ā)送端綁定一個routing_key1, queue中綁定若干個routing_key2, 若key1與key2相等，或者key1在key2中，則消息就會發(fā)送到這個queue中，再由相應(yīng)的消費者去queue中取數(shù)據(jù)。
publisher.py：
from base import channel, connectionchannel.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.queuechannel.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, connectionchannel.exchange_declare(exchange="direct_test", exchange_type="direct")result = channel.queue_declare(exclusive=True)queue_name = result.method.queuechannel.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.queuekey_list = ["info", "warning"]for key in key_list: channel.queue_bind( exchange="direct_test", queue=queue_name, # 一個queue同時綁定多個key，有一個key滿足條件時就可以收到數(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.pypython consumer01.pypython consumer02.pypython consumer03.py
結(jié)果:
consumer01.py: body:b'hello'
consumer02.py沒收到結(jié)果
consumer03.py: body:b'hello'
topic模式不是太好理解，我的理解如下:
對于發(fā)送端綁定的routing_key1，queue綁定若干個routing_key2；若routing_key1滿足任意一個routing_key2，則該消息就會通過exchange發(fā)送到這個queue中，然后由接收端從queue中取出其實就是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 sysfrom base import channel, connection# 聲明exchangechannel.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.queuechannel.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.queuechannel.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.pypython consumer02.py
結(jié)果:
consumer01.py的結(jié)果: body:b'this is a topic test'
consumer02.py的結(jié)果: body:b'this is a topic test'
說明通過綁定相應(yīng)的routing_key，兩個消費者都收到了消息
將publisher.py的routing_key改成"mysql.info"
再此執(zhí)行:
python publisher02.py "this is a topic test"python consumer01.pypython consumer02.py
結(jié)果:
consumer01.py沒收到結(jié)果
consumer02.py的結(jié)果: body:b'this is a topic test'
通過這個例子我們就能明白topic的運行方式了。
以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持VEVB武林網(wǎng)。
新聞熱點
疑難解答
圖片精選
