RabbitMQ可以當(dāng)做一個(gè)消息代理，它的核心原理非常簡(jiǎn)單：即接收和發(fā)送消息，可以把它想象成一個(gè)郵局：我們把信件放入郵箱，郵遞員就會(huì)把信件投遞到你的收件人處，RabbitMQ就是一個(gè)郵箱、郵局、投遞員功能綜合體，整個(gè)過(guò)程就是：郵箱接收信件，郵局轉(zhuǎn)發(fā)信件，投遞員投遞信件到達(dá)收件人處。

RabbitMQ和郵局的主要區(qū)別就是RabbitMQ接收、存儲(chǔ)和發(fā)送的是二進(jìn)制數(shù)據(jù)----消息。

rabbitmq基本管理命令：

一步啟動(dòng)Erlang node和Rabbit應(yīng)用：sudo rabbitmq-server

在后臺(tái)啟動(dòng)Rabbit node：sudo rabbitmq-server -detached

關(guān)閉整個(gè)節(jié)點(diǎn)（包括應(yīng)用）：sudo rabbitmqctl stop

add_user <UserName> <Password>
delete_user <UserName>
change_password <UserName> <NewPassword>
list_users
add_vhost <VHostPath>
delete_vhost <VHostPath>
list_vhosts
set_permissions [-p <VHostPath>] <UserName> <Regexp> <Regexp> <Regexp>
clear_permissions [-p <VHostPath>] <UserName>
list_permissions [-p <VHostPath>]
list_user_permissions <UserName>
list_queues [-p <VHostPath>] [<QueueInfoItem> ...]
list_exchanges [-p <VHostPath>] [<ExchangeInfoItem> ...]
list_bindings [-p <VHostPath>]
list_connections [<ConnectionInfoItem> ...]

Demo:

producer.py

 #!/usr/bin/env python
 # -*- coding: utf_ -*-
 # Date: 年月日
 # Author:蔚藍(lán)行
 # 博客 http://www.cnblogs.com/duanv/
 import pika
 import sys
 #創(chuàng)建連接connection到localhost
 con = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
 #創(chuàng)建虛擬連接channel
 cha = con.channel()
 #創(chuàng)建隊(duì)列anheng,durable參數(shù)為真時(shí)，隊(duì)列將持久化；exclusive為真時(shí)，建立臨時(shí)隊(duì)列
 result=cha.queue_declare(queue='anheng',durable=True,exclusive=False)
 #創(chuàng)建名為yanfa,類型為fanout的exchange，其他類型還有direct和topic，如果指定durable為真，exchange將持久化
 cha.exchange_declare(durable=False,
           exchange='yanfa',
           type='direct',)
 #綁定exchange和queue,result.method.queue獲取的是隊(duì)列名稱
 cha.queue_bind(exchange='yanfa', 
        queue=result.method.queue,
        routing_key='',) 
 #公平分發(fā)，使每個(gè)consumer在同一時(shí)間最多處理一個(gè)message，收到ack前，不會(huì)分配新的message
 cha.basic_qos(prefetch_count=)
 #發(fā)送信息到隊(duì)列‘a(chǎn)nheng'
 message = ' '.join(sys.argv[:])
 #消息持久化指定delivery_mode=；
 cha.basic_publish(exchange='',
          routing_key='anheng',
          body=message,
          properties=pika.BasicProperties(
           delivery_mode = ,
         ))
 print '[x] Sent %r' % (message,)
 #關(guān)閉連接
 con.close()

consumer.py

 #!/usr/bin/env python
 # -*- coding: utf_ -*-
 # Date: 年月日
 # Author:蔚藍(lán)行
 # 博客 http://www.cnblogs.com/duanv/
 import pika
 #建立連接connection到localhost
 con = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
 #創(chuàng)建虛擬連接channel
 cha = con.channel()
 #創(chuàng)建隊(duì)列anheng
 result=cha.queue_declare(queue='anheng',durable=True)
 #創(chuàng)建名為yanfa,類型為fanout的交換機(jī)，其他類型還有direct和topic
 cha.exchange_declare(durable=False,
           exchange='yanfa', 
           type='direct',)
 #綁定exchange和queue,result.method.queue獲取的是隊(duì)列名稱
 cha.queue_bind(exchange='yanfa',
        queue=result.method.queue,
        routing_key='',)
 #公平分發(fā)，使每個(gè)consumer在同一時(shí)間最多處理一個(gè)message，收到ack前，不會(huì)分配新的message
 cha.basic_qos(prefetch_count=)
 print ' [*] Waiting for messages. To exit press CTRL+C'
 #定義回調(diào)函數(shù)
 def callback(ch, method, properties, body):
   print " [x] Received %r" % (body,)
   ch.basic_ack(delivery_tag = method.delivery_tag)
 cha.basic_consume(callback,
          queue='anheng',
          no_ack=False,)
 cha.start_consuming()

一、概念：

Connection： 一個(gè)TCP的連接。Producer和Consumer都是通過(guò)TCP連接到RabbitMQ Server的。程序的起始處就是建立這個(gè)TCP連接。

Channels： 虛擬連接。建立在上述的TCP連接中。數(shù)據(jù)流動(dòng)都是在Channel中進(jìn)行的。一般情況是程序起始建立TCP連接，第二步就是建立這個(gè)Channel。

二、隊(duì)列：

首先建立一個(gè)Connection，然后建立Channels，在channel上建立隊(duì)列

建立時(shí)指定durable參數(shù)為真，隊(duì)列將持久化；指定exclusive為真，隊(duì)列為臨時(shí)隊(duì)列，關(guān)閉consumer后該隊(duì)列將不再存在，一般情況下建立臨時(shí)隊(duì)列并不指定隊(duì)列名稱，rabbitmq將隨機(jī)起名，通過(guò)result.method.queue來(lái)獲取隊(duì)列名：

result = channel.queue_declare(exclusive=True)

result.method.queue

區(qū)別：durable是隊(duì)列持久化與否，如果為真，隊(duì)列將在rabbitmq服務(wù)重啟后仍存在，如果為假，rabbitmq服務(wù)重啟前不會(huì)消失，與consumer關(guān)閉與否無(wú)關(guān)；

而exclusive是建立臨時(shí)隊(duì)列，當(dāng)consumer關(guān)閉后，該隊(duì)列就會(huì)被刪除

三、exchange和bind

Exchange中durable參數(shù)指定exchange是否持久化，exchange參數(shù)指定exchange名稱，type指定exchange類型。Exchange類型有direct，fanout和topic。

Bind是將exchange與queue進(jìn)行關(guān)聯(lián)，exchange參數(shù)和queue參數(shù)分別指定要進(jìn)行bind的exchange和queue，routing_key為可選參數(shù)。

Exchange的三種模式：

Direct：

任何發(fā)送到Direct Exchange的消息都會(huì)被轉(zhuǎn)發(fā)到routing_key中指定的Queue

1.一般情況可以使用rabbitMQ自帶的Exchange：””(該Exchange的名字為空字符串)；

2.這種模式下不需要將Exchange進(jìn)行任何綁定(bind)操作；

3.消息傳遞時(shí)需要一個(gè)“routing_key”，可以簡(jiǎn)單的理解為要發(fā)送到的隊(duì)列名字；

4.如果vhost中不存在routing_key中指定的隊(duì)列名，則該消息會(huì)被拋棄。

Demo中雖然聲明了一個(gè)exchange='yanfa'和queue='anheng'的bind，但是在后面發(fā)送消息時(shí)并沒(méi)有使用該exchange和bind，而是采用了direct的模式，沒(méi)有指定exchange，而是指定了routing_key的名稱為隊(duì)列名，消息將發(fā)送到指定隊(duì)列。

如果一個(gè)exchange 聲明為direct，并且bind中指定了routing_key,那么發(fā)送消息時(shí)需要同時(shí)指明該exchange和routing_key.

Fanout:

任何發(fā)送到Fanout Exchange的消息都會(huì)被轉(zhuǎn)發(fā)到與該Exchange綁定(Binding)的所有Queue上

1.可以理解為路由表的模式

2.這種模式不需要routing_key

3.這種模式需要提前將Exchange與Queue進(jìn)行綁定，一個(gè)Exchange可以綁定多個(gè)Queue，一個(gè)Queue可以同多個(gè)Exchange進(jìn)行綁定。

4.如果接受到消息的Exchange沒(méi)有與任何Queue綁定，則消息會(huì)被拋棄。

Demo中創(chuàng)建了一個(gè)將一個(gè)exchange和一個(gè)queue進(jìn)行fanout類型的bind.但是發(fā)送信息時(shí)沒(méi)有用到它，如果要用到它，只要在發(fā)送消息時(shí)指定該exchange的名稱即可，該exchange就會(huì)將消息發(fā)送到所有和它bind的隊(duì)列中。在fanout模式下，指定的routing_key是無(wú)效的 。

Topic：

任何發(fā)送到Topic Exchange的消息都會(huì)被轉(zhuǎn)發(fā)到所有關(guān)心routing_key中指定話題的Queue上

1.這種模式較為復(fù)雜，簡(jiǎn)單來(lái)說(shuō)，就是每個(gè)隊(duì)列都有其關(guān)心的主題，所有的消息都帶有一個(gè)“標(biāo)題”(routing_key)，Exchange會(huì)將消息轉(zhuǎn)發(fā)到所有關(guān)注主題能與routing_key模糊匹配的隊(duì)列。

2.這種模式需要routing_key，也許要提前綁定Exchange與Queue。

3.在進(jìn)行綁定時(shí)，要提供一個(gè)該隊(duì)列關(guān)心的主題，如“#.log.#”表示該隊(duì)列關(guān)心所有涉及l(fā)og的消息(一個(gè)routing_key為”MQ.log.error”的消息會(huì)被轉(zhuǎn)發(fā)到該隊(duì)列)。

4.“#”表示0個(gè)或若干個(gè)關(guān)鍵字，“*”表示一個(gè)關(guān)鍵字。如“l(fā)og.*”能與“l(fā)og.warn”匹配，無(wú)法與“l(fā)og.warn.timeout”匹配；但是“l(fā)og.#”能與上述兩者匹配。

5.同樣，如果Exchange沒(méi)有發(fā)現(xiàn)能夠與routing_key匹配的Queue，則會(huì)拋棄此消息。

四、任務(wù)分發(fā)

1.Rabbitmq的任務(wù)是循環(huán)分發(fā)的，如果開(kāi)啟兩個(gè)consumer，producer發(fā)送的信息是輪流發(fā)送到兩個(gè)consume的。

2.在producer端使用cha.basic_publish()來(lái)發(fā)送消息，其中body參數(shù)就是要發(fā)送的消息，properties=pika.BasicProperties(delivery_mode = 2,)啟用消息持久化，可以防止RabbitMQ Server 重啟或者crash引起的數(shù)據(jù)丟失。

3.在接收端使用cha.basic_consume()無(wú)限循環(huán)監(jiān)聽(tīng)，如果設(shè)置no-ack參數(shù)為真，每次Consumer接到數(shù)據(jù)后，而不管是否處理完成，RabbitMQ Server會(huì)立即把這個(gè)Message標(biāo)記為完成，然后從queue中刪除了。為了保證數(shù)據(jù)不被丟失，RabbitMQ支持消息確認(rèn)機(jī)制，即acknowledgments。為了保證數(shù)據(jù)能被正確處理而不僅僅是被Consumer收到，那么我們不能采用no-ack。而應(yīng)該是在處理完數(shù)據(jù)后發(fā)送ack。

在處理數(shù)據(jù)后發(fā)送的ack，就是告訴RabbitMQ數(shù)據(jù)已經(jīng)被接收，處理完成，RabbitMQ可以去安全的刪除它了。如果Consumer退出了但是沒(méi)有發(fā)送ack，那么RabbitMQ就會(huì)把這個(gè)Message發(fā)送到下一個(gè)Consumer。這樣就保證了在Consumer異常退出的情況下數(shù)據(jù)也不會(huì)丟失。

這里并沒(méi)有用到超時(shí)機(jī)制。RabbitMQ僅僅通過(guò)Consumer的連接中斷來(lái)確認(rèn)該Message并沒(méi)有被正確處理。也就是說(shuō)，RabbitMQ給了Consumer足夠長(zhǎng)的時(shí)間來(lái)做數(shù)據(jù)處理。

Demo的callback方法中ch.basic_ack(delivery_tag = method.delivery_tag)告訴rabbitmq消息已經(jīng)正確處理。如果沒(méi)有這條代碼，Consumer退出時(shí)，Message會(huì)重新分發(fā)。然后RabbitMQ會(huì)占用越來(lái)越多的內(nèi)存，由于RabbitMQ會(huì)長(zhǎng)時(shí)間運(yùn)行，因此這個(gè)“內(nèi)存泄漏”是致命的。去調(diào)試這種錯(cuò)誤，可以通過(guò)一下命令打印un-acked Messages：

sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged

4.公平分發(fā)：設(shè)置cha.basic_qos(prefetch_count=1)，這樣RabbitMQ就會(huì)使得每個(gè)Consumer在同一個(gè)時(shí)間點(diǎn)最多處理一個(gè)Message。換句話說(shuō)，在接收到該Consumer的ack前，他它不會(huì)將新的Message分發(fā)給它。

五、注意：

生產(chǎn)者和消費(fèi)者都應(yīng)該聲明建立隊(duì)列，網(wǎng)上教程上說(shuō)第二次創(chuàng)建如果參數(shù)和第一次不一樣，那么該操作雖然成功，但是queue的屬性并不會(huì)被修改。

可能因?yàn)榘姹締?wèn)題，在我的測(cè)試中如果第二次聲明建立的隊(duì)列屬性和第一次不完全相同，將報(bào)類似這種錯(cuò)406, "PRECONDITION_FAILED - parameters for queue 'anheng' in vhost '/' not equivalent"

如果是exchange第二次創(chuàng)建屬性不同，將報(bào)這種錯(cuò)406, "PRECONDITION_FAILED - cannot redeclare exchange 'yanfa' in vhost '/' with different type, durable, internal or autodelete value"

如果第一次聲明建立隊(duì)列也出現(xiàn)這個(gè)錯(cuò)誤，說(shuō)明之前存在名字相同的隊(duì)列且本次聲明的某些屬性和之前聲明不同，可通過(guò)命令sudo rabbitmqctl list_queues查看當(dāng)前有哪些隊(duì)列。解決方法是聲明建立另一名稱的隊(duì)列或刪除原有隊(duì)列，如果原有隊(duì)列是非持久化的，可通過(guò)重啟rabbitmq服務(wù)刪除原有隊(duì)列，如果原有隊(duì)列是持久化的，只能刪除它所在的vhost,然后再重建vhost,再設(shè)置vhost的權(quán)限（先確認(rèn)該vhost中沒(méi)有其他有用隊(duì)列）。

sudo rabbitmqctl delete_vhost /
sudo rabbitmqctl add_vhost /
sudo rabbitmqctl set_permissions -p / username '.*' '.*' '.*'

以上內(nèi)容是小編給大家介紹的利用Python學(xué)習(xí)RabbitMQ消息隊(duì)列，希望大家喜歡。

最新評(píng)論