最近在研究redis做消息隊列時，順便看了一下RabbitMQ做消息隊列的實現(xiàn)。以下是總結的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, 不聲明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會隨機分配一個名字, 消息處理完成后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中，再由相應的消費者去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()