RabbitMQ與Redis對(duì)比

​ RabbitMQ是一種比較流行的消息中間件，之前我一直使用redis作為消息中間件，但是生產(chǎn)環(huán)境比較推薦RabbitMQ來替代Redis，所以我去查詢了一些RabbitMQ的資料。相比于Redis，RabbitMQ優(yōu)點(diǎn)很多，比如：

具有消息消費(fèi)確認(rèn)機(jī)制 隊(duì)列，消息，都可以選擇是否持久化，粒度更小、更靈活。 可以實(shí)現(xiàn)負(fù)載均衡

RabbitMQ應(yīng)用場(chǎng)景

異步處理：比如用戶注冊(cè)時(shí)的確認(rèn)郵件、短信等交由rabbitMQ進(jìn)行異步處理 應(yīng)用解耦：比如收發(fā)消息雙方可以使用消息隊(duì)列，具有一定的緩沖功能 流量削峰：一般應(yīng)用于秒殺活動(dòng)，可以控制用戶人數(shù)，也可以降低流量 日志處理：將info、warning、error等不同的記錄分開存儲(chǔ)

RabbitMQ消息模型

​ 這里使用 Python 的 pika 這個(gè)庫來實(shí)現(xiàn)RabbitMQ中常見的6種消息模型。沒有的可以先安裝：

pip install pika

1.單生產(chǎn)單消費(fèi)模型：即完成基本的一對(duì)一消息轉(zhuǎn)發(fā)。

# 生產(chǎn)者代碼import pikacredentials = pika.PlainCredentials(’chuan’, ’123’) # mq用戶名和密碼，沒有則需要自己創(chuàng)建# 虛擬隊(duì)列需要指定參數(shù) virtual_host，如果是默認(rèn)的可以不填。connection = pika.BlockingConnection(pika.ConnectionParameters(host=’localhost’,port=5672,virtual_host=’/’,credentials=credentials))# 建立rabbit協(xié)議的通道channel = connection.channel()# 聲明消息隊(duì)列，消息將在這個(gè)隊(duì)列傳遞，如不存在，則創(chuàng)建。durable指定隊(duì)列是否持久化channel.queue_declare(queue=’python-test’, durable=False)# message不能直接發(fā)送給queue，需經(jīng)exchange到達(dá)queue，此處使用以空字符串標(biāo)識(shí)的默認(rèn)的exchange# 向隊(duì)列插入數(shù)值 routing_key是隊(duì)列名channel.basic_publish(exchange=’’, routing_key=’python-test’, body=’Hello world！2’)# 關(guān)閉與rabbitmq server的連接connection.close()

# 消費(fèi)者代碼import pikacredentials = pika.PlainCredentials(’chuan’, ’123’)# BlockingConnection:同步模式connection = pika.BlockingConnection(pika.ConnectionParameters(host=’localhost’,port=5672,virtual_host=’/’, credentials=credentials))channel = connection.channel()# 申明消息隊(duì)列。當(dāng)不確定生產(chǎn)者和消費(fèi)者哪個(gè)先啟動(dòng)時(shí)，可以兩邊重復(fù)聲明消息隊(duì)列。channel.queue_declare(queue=’python-test’, durable=False)# 定義一個(gè)回調(diào)函數(shù)來處理消息隊(duì)列中的消息，這里是打印出來def callback(ch, method, properties, body): # 手動(dòng)發(fā)送確認(rèn)消息 ch.basic_ack(delivery_tag=method.delivery_tag) print(body.decode()) # 告訴生產(chǎn)者，消費(fèi)者已收到消息# 告訴rabbitmq，用callback來接收消息# 默認(rèn)情況下是要對(duì)消息進(jìn)行確認(rèn)的，以防止消息丟失。# 此處將auto_ack明確指明為True，不對(duì)消息進(jìn)行確認(rèn)。channel.basic_consume(’python-test’, on_message_callback=callback) # auto_ack=True) # 自動(dòng)發(fā)送確認(rèn)消息# 開始接收信息，并進(jìn)入阻塞狀態(tài)，隊(duì)列里有信息才會(huì)調(diào)用callback進(jìn)行處理channel.start_consuming()

2.消息分發(fā)模型：多個(gè)收聽者監(jiān)聽一個(gè)隊(duì)列。

# 生產(chǎn)者代碼import pikacredentials = pika.PlainCredentials(’chuan’, ’123’) # mq用戶名和密碼# 虛擬隊(duì)列需要指定參數(shù) virtual_host，如果是默認(rèn)的可以不填。connection = pika.BlockingConnection(pika.ConnectionParameters(host=’localhost’,port=5672,virtual_host=’/’,credentials=credentials))# 建立rabbit協(xié)議的通道channel = connection.channel()# 聲明消息隊(duì)列，消息將在這個(gè)隊(duì)列傳遞，如不存在，則創(chuàng)建。durable指定隊(duì)列是否持久化。確保沒有確認(rèn)的消息不會(huì)丟失channel.queue_declare(queue=’rabbitmqtest’, durable=True)# message不能直接發(fā)送給queue，需經(jīng)exchange到達(dá)queue，此處使用以空字符串標(biāo)識(shí)的默認(rèn)的exchange# 向隊(duì)列插入數(shù)值 routing_key是隊(duì)列名# basic_publish的properties參數(shù)指定message的屬性。此處delivery_mode=2指明message為持久的for i in range(10): channel.basic_publish(exchange=’’, routing_key=’python-test’, body=’Hello world！%s’ % i, properties=pika.BasicProperties(delivery_mode=2))# 關(guān)閉與rabbitmq server的連接connection.close()

# 消費(fèi)者代碼,consume1與consume2import pikaimport timecredentials = pika.PlainCredentials(’chuan’, ’123’)# BlockingConnection:同步模式connection = pika.BlockingConnection(pika.ConnectionParameters(host=’localhost’,port=5672,virtual_host=’/’,credentials=credentials))channel = connection.channel()# 申明消息隊(duì)列。當(dāng)不確定生產(chǎn)者和消費(fèi)者哪個(gè)先啟動(dòng)時(shí)，可以兩邊重復(fù)聲明消息隊(duì)列。channel.queue_declare(queue=’rabbitmqtest’, durable=True)# 定義一個(gè)回調(diào)函數(shù)來處理消息隊(duì)列中的消息，這里是打印出來def callback(ch, method, properties, body): # 手動(dòng)發(fā)送確認(rèn)消息 time.sleep(10) print(body.decode()) # 告訴生產(chǎn)者，消費(fèi)者已收到消息 ch.basic_ack(delivery_tag=method.delivery_tag)# 如果該消費(fèi)者的channel上未確認(rèn)的消息數(shù)達(dá)到了prefetch_count數(shù)，則不向該消費(fèi)者發(fā)送消息channel.basic_qos(prefetch_count=1)# 告訴rabbitmq，用callback來接收消息# 默認(rèn)情況下是要對(duì)消息進(jìn)行確認(rèn)的，以防止消息丟失。# 此處將no_ack明確指明為True，不對(duì)消息進(jìn)行確認(rèn)。channel.basic_consume(’python-test’, on_message_callback=callback) # auto_ack=True) # 自動(dòng)發(fā)送確認(rèn)消息# 開始接收信息，并進(jìn)入阻塞狀態(tài)，隊(duì)列里有信息才會(huì)調(diào)用callback進(jìn)行處理channel.start_consuming()

3.fanout消息訂閱模式：生產(chǎn)者將消息發(fā)送到Exchange，Exchange再轉(zhuǎn)發(fā)到與之綁定的Queue中，每個(gè)消費(fèi)者再到自己的Queue中取消息。

# 生產(chǎn)者代碼import pikacredentials = pika.PlainCredentials(’chuan’, ’123’) # mq用戶名和密碼# 虛擬隊(duì)列需要指定參數(shù) virtual_host，如果是默認(rèn)的可以不填。connection = pika.BlockingConnection(pika.ConnectionParameters(host=’localhost’,port=5672,virtual_host=’/’,credentials=credentials))# 建立rabbit協(xié)議的通道channel = connection.channel()# fanout: 所有綁定到此exchange的queue都可以接收消息（實(shí)時(shí)廣播）# direct: 通過routingKey和exchange決定的那一組的queue可以接收消息（有選擇接受）# topic： 所有符合routingKey(此時(shí)可以是一個(gè)表達(dá)式)的routingKey所bind的queue可以接收消息（更細(xì)致的過濾）channel.exchange_declare(’logs’, exchange_type=’fanout’)#因?yàn)槭莊anout廣播類型的exchange，這里無需指定routing_keyfor i in range(10): channel.basic_publish(exchange=’logs’, routing_key=’’, body=’Hello world！%s’ % i)# 關(guān)閉與rabbitmq server的連接connection.close()

import pikacredentials = pika.PlainCredentials(’chuan’, ’123’)# BlockingConnection:同步模式connection = pika.BlockingConnection(pika.ConnectionParameters(host=’localhost’,port=5672,virtual_host=’/’,credentials=credentials))channel = connection.channel()#作為好的習(xí)慣，在producer和consumer中分別聲明一次以保證所要使用的exchange存在channel.exchange_declare(exchange=’logs’, exchange_type=’fanout’)# 隨機(jī)生成一個(gè)新的空的queue，將exclusive置為True，這樣在consumer從RabbitMQ斷開后會(huì)刪除該queue# 是排他的。result = channel.queue_declare(’’, exclusive=True)# 用于獲取臨時(shí)queue的namequeue_name = result.method.queue# exchange與queue之間的關(guān)系成為binding# binding告訴exchange將message發(fā)送該哪些queuechannel.queue_bind(exchange=’logs’, queue=queue_name)# 定義一個(gè)回調(diào)函數(shù)來處理消息隊(duì)列中的消息，這里是打印出來def callback(ch, method, properties, body): # 手動(dòng)發(fā)送確認(rèn)消息 print(body.decode()) # 告訴生產(chǎn)者，消費(fèi)者已收到消息 #ch.basic_ack(delivery_tag=method.delivery_tag)# 如果該消費(fèi)者的channel上未確認(rèn)的消息數(shù)達(dá)到了prefetch_count數(shù)，則不向該消費(fèi)者發(fā)送消息channel.basic_qos(prefetch_count=1)# 告訴rabbitmq，用callback來接收消息# 默認(rèn)情況下是要對(duì)消息進(jìn)行確認(rèn)的，以防止消息丟失。# 此處將no_ack明確指明為True，不對(duì)消息進(jìn)行確認(rèn)。channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True) # 自動(dòng)發(fā)送確認(rèn)消息# 開始接收信息，并進(jìn)入阻塞狀態(tài)，隊(duì)列里有信息才會(huì)調(diào)用callback進(jìn)行處理channel.start_consuming()

4.direct路由模式：此時(shí)生產(chǎn)者發(fā)送消息時(shí)需要指定RoutingKey，即路由Key，Exchange接收到消息時(shí)轉(zhuǎn)發(fā)到與RoutingKey相匹配的隊(duì)列中。

# 生產(chǎn)者代碼，測(cè)試命令可以使用：python produce.py error 404errorimport pikaimport sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host=’localhost’))channel = connection.channel()# 聲明一個(gè)名為direct_logs的direct類型的exchange# direct類型的exchangechannel.exchange_declare(exchange=’direct_logs’, exchange_type=’direct’)# 從命令行獲取basic_publish的配置參數(shù)severity = sys.argv[1] if len(sys.argv) > 1 else ’info’message = ’ ’.join(sys.argv[2:]) or ’Hello World!’# 向名為direct_logs的exchage按照設(shè)置的routing_key發(fā)送messagechannel.basic_publish(exchange=’direct_logs’, routing_key=severity, body=message)print(' [x] Sent %r:%r' % (severity, message))connection.close()

# 消費(fèi)者代碼，測(cè)試可以使用：python consume.py errorimport pikaimport sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host=’localhost’))channel = connection.channel()# 聲明一個(gè)名為direct_logs類型為direct的exchange# 同時(shí)在producer和consumer中聲明exchage或queue是個(gè)好習(xí)慣，以保證其存在channel.exchange_declare(exchange=’direct_logs’, exchange_type=’direct’)result = channel.queue_declare(’’, exclusive=True)queue_name = result.method.queue# 從命令行獲取參數(shù):routing_keyseverities = sys.argv[1:]if not severities: print(sys.stderr, 'Usage: %s [info] [warning] [error]' % (sys.argv[0],)) sys.exit(1)for severity in severities: # exchange和queue之間的binding可接受routing_key參數(shù) # fanout類型的exchange直接忽略該參數(shù)。direct類型的exchange精確匹配該關(guān)鍵字進(jìn)行message路由 # 一個(gè)消費(fèi)者可以綁定多個(gè)routing_key # Exchange就是根據(jù)這個(gè)RoutingKey和當(dāng)前Exchange所有綁定的BindingKey做匹配， # 如果滿足要求，就往BindingKey所綁定的Queue發(fā)送消息 channel.queue_bind(exchange=’direct_logs’, queue=queue_name, routing_key=severity)def callback(ch, method, properties, body): print(' [x] %r:%r' % (method.routing_key, body,))channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)channel.start_consuming()

5.topic匹配模式：更細(xì)致的分組，允許在RoutingKey中使用匹配符。

*：匹配一個(gè)單詞 #：匹配0個(gè)或多個(gè)單詞

# 生產(chǎn)者代碼，基本不變，只需將exchange_type改為topic（測(cè)試：python produce.py rabbitmq.red # red color is my favoriteimport pikaimport sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host=’localhost’))channel = connection.channel()# 聲明一個(gè)名為direct_logs的direct類型的exchange# direct類型的exchangechannel.exchange_declare(exchange=’topic_logs’, exchange_type=’topic’)# 從命令行獲取basic_publish的配置參數(shù)severity = sys.argv[1] if len(sys.argv) > 1 else ’info’message = ’ ’.join(sys.argv[2:]) or ’Hello World!’# 向名為direct_logs的exchange按照設(shè)置的routing_key發(fā)送messagechannel.basic_publish(exchange=’topic_logs’, routing_key=severity, body=message)print(' [x] Sent %r:%r' % (severity, message))connection.close()

# 消費(fèi)者代碼，（測(cè)試：python consume.py *.red）import pikaimport sysconnection = pika.BlockingConnection(pika.ConnectionParameters(host=’localhost’))channel = connection.channel()# 聲明一個(gè)名為direct_logs類型為direct的exchange# 同時(shí)在producer和consumer中聲明exchage或queue是個(gè)好習(xí)慣，以保證其存在channel.exchange_declare(exchange=’topic_logs’, exchange_type=’topic’)result = channel.queue_declare(’’, exclusive=True)queue_name = result.method.queue# 從命令行獲取參數(shù):routing_keyseverities = sys.argv[1:]if not severities: print(sys.stderr, 'Usage: %s [info] [warning] [error]' % (sys.argv[0],)) sys.exit(1)for severity in severities: # exchange和queue之間的binding可接受routing_key參數(shù) # fanout類型的exchange直接忽略該參數(shù)。direct類型的exchange精確匹配該關(guān)鍵字進(jìn)行message路由 # 一個(gè)消費(fèi)者可以綁定多個(gè)routing_key # Exchange就是根據(jù)這個(gè)RoutingKey和當(dāng)前Exchange所有綁定的BindingKey做匹配， # 如果滿足要求，就往BindingKey所綁定的Queue發(fā)送消息 channel.queue_bind(exchange=’topic_logs’, queue=queue_name, routing_key=severity)def callback(ch, method, properties, body): print(' [x] %r:%r' % (method.routing_key, body,))channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)channel.start_consuming()

6.RPC遠(yuǎn)程過程調(diào)用：客戶端與服務(wù)器之間是完全解耦的，即兩端既是消息的發(fā)送者也是接受者。

# 生產(chǎn)者代碼import pikaimport uuid# 在一個(gè)類中封裝了connection建立、queue聲明、consumer配置、回調(diào)函數(shù)等class FibonacciRpcClient(object): def __init__(self): # 建立到RabbitMQ Server的connection self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=’localhost’)) self.channel = self.connection.channel() # 聲明一個(gè)臨時(shí)的回調(diào)隊(duì)列 result = self.channel.queue_declare(’’, exclusive=True) self._queue = result.method.queue # 此處client既是producer又是consumer，因此要配置consume參數(shù) # 這里的指明從client自己創(chuàng)建的臨時(shí)隊(duì)列中接收消息 # 并使用on_response函數(shù)處理消息 # 不對(duì)消息進(jìn)行確認(rèn) self.channel.basic_consume(queue=self._queue, on_message_callback=self.on_response, auto_ack=True) self.response = None self.corr_id = None # 定義回調(diào)函數(shù) # 比較類的corr_id屬性與props中corr_id屬性的值 # 若相同則response屬性為接收到的message def on_response(self, ch, method, props, body): if self.corr_id == props.correlation_id: self.response = body def call(self, n): # 初始化response和corr_id屬性 self.corr_id = str(uuid.uuid4()) # 使用默認(rèn)exchange向server中定義的rpc_queue發(fā)送消息 # 在properties中指定replay_to屬性和correlation_id屬性用于告知遠(yuǎn)程server # correlation_id屬性用于匹配request和response self.channel.basic_publish(exchange=’’, routing_key=’rpc_queue’, properties=pika.BasicProperties( reply_to=self._queue, correlation_id=self.corr_id, ), # message需為字符串 body=str(n)) while self.response is None: self.connection.process_data_events() return int(self.response)# 生成類的實(shí)例fibonacci_rpc = FibonacciRpcClient()print(' [x] Requesting fib(30)')# 調(diào)用實(shí)例的call方法response = fibonacci_rpc.call(30)print(' [.] Got %r' % response)

# 消費(fèi)者代碼，這里以生成斐波那契數(shù)列為例import pika# 建立到達(dá)RabbitMQ Server的connectionconnection = pika.BlockingConnection(pika.ConnectionParameters(host=’localhost’))channel = connection.channel()# 聲明一個(gè)名為rpc_queue的queuechannel.queue_declare(queue=’rpc_queue’)# 計(jì)算指定數(shù)字的斐波那契數(shù)def fib(n): if n == 0: return 0 elif n == 1: return 1 else: return fib(n - 1) + fib(n - 2)# 回調(diào)函數(shù)，從queue接收到message后調(diào)用該函數(shù)進(jìn)行處理def on_request(ch, method, props, body): # 由message獲取要計(jì)算斐波那契數(shù)的數(shù)字 n = int(body) print(' [.] fib(%s)' % n) # 調(diào)用fib函數(shù)獲得計(jì)算結(jié)果 response = fib(n) # exchage為空字符串則將message發(fā)送個(gè)到routing_key指定的queue # 這里queue為回調(diào)函數(shù)參數(shù)props中reply_ro指定的queue # 要發(fā)送的message為計(jì)算所得的斐波那契數(shù) # properties中correlation_id指定為回調(diào)函數(shù)參數(shù)props中co的rrelation_id # 最后對(duì)消息進(jìn)行確認(rèn) ch.basic_publish(exchange=’’, routing_key=props.reply_to, properties=pika.BasicProperties(correlation_id=props.correlation_id), body=str(response)) ch.basic_ack(delivery_tag=method.delivery_tag)# 只有consumer已經(jīng)處理并確認(rèn)了上一條message時(shí)queue才分派新的message給它c(diǎn)hannel.basic_qos(prefetch_count=1)# 設(shè)置consumeer參數(shù)，即從哪個(gè)queue獲取消息使用哪個(gè)函數(shù)進(jìn)行處理，是否對(duì)消息進(jìn)行確認(rèn)channel.basic_consume(queue=’rpc_queue’, on_message_callback=on_request)print(' [x] Awaiting RPC requests')# 開始接收并處理消息channel.start_consuming()

到此這篇關(guān)于Python實(shí)現(xiàn)RabbitMQ6種消息模型的示例代碼的文章就介紹到這了,更多相關(guān)Python RabbitMQ消息模型 內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！