摘要：最近在研究做消息隊列時，順便看了一下做消息隊列的實現(xiàn)。遠程連接時需要認證實例化連接對象實例化鏈接參數(shù)對象創(chuàng)建新的通道模式向綁定到指定的中發(fā)送消息，消費者從中取出數(shù)據(jù)，類似于廣播模式發(fā)布訂閱模式。

最近在研究redis做消息隊列時，順便看了一下RabbitMQ做消息隊列的實現(xiàn)。以下是總結(jié)的RabbitMQ中三種exchange模式的實現(xiàn)，分別是fanout, direct和topic。

base.py：

import pika
        
        
# 獲取認證對象，參數(shù)是用戶名、密碼。遠程連接時需要認證
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, 不聲明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會隨機分配一個名字, 消息處理完成后queue會自動刪除
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ā)送端綁定一個routing_key1, queue中綁定若干個routing_key2, 若key1與key2相等，或者key1在key2中，則消息就會發(fā)送到這個queue中，再由相應(yīng)的消費者去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,
        # 一個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.py
python consumer01.py
python consumer02.py
python 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模式的擴展。

綁定方式:
發(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"

說明通過綁定相應(yīng)的routing_key，兩個消費者都收到了消息

將publisher.py的routing_key改成"mysql.info"
再此執(zhí)行:

python publisher02.py "this is a topic test"
python consumer01.py
python consumer02.py

結(jié)果:

consumer01.py沒收到結(jié)果
consumer02.py的結(jié)果: body:b"this is a topic test"

通過這個例子我們就能明白topic的運行方式了。

參考自： https://blog.csdn.net/fgf00/a...

今天就說到這里，如有問題，歡迎交流指正！