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

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

rabbitmq基本管理命令：

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

在后臺啟動Rabbit node：sudo rabbitmq-server -detached

關(guān)閉整個節(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ù)為真時，隊(duì)列將持久化；exclusive為真時，建立臨時隊(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ā)，使每個consumer在同一時間最多處理一個message，收到ack前，不會分配新的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ā)，使每個consumer在同一時間最多處理一個message，收到ack前，不會分配新的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： 一個TCP的連接。Producer和Consumer都是通過TCP連接到RabbitMQ Server的。程序的起始處就是建立這個TCP連接。

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

二、隊(duì)列：

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

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

result = channel.queue_declare(exclusive=True)

result.method.queue

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

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

三、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的消息都會被轉(zhuǎn)發(fā)到routing_key中指定的Queue

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

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

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

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

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

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

Fanout:

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

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

2.這種模式不需要routing_key

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

4.如果接受到消息的Exchange沒有與任何Queue綁定，則消息會被拋棄。

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

Topic：

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

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

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

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

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

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

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

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

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

3.在接收端使用cha.basic_consume()無限循環(huán)監(jiān)聽，如果設(shè)置no-ack參數(shù)為真，每次Consumer接到數(shù)據(jù)后，而不管是否處理完成，RabbitMQ Server會立即把這個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退出了但是沒有發(fā)送ack，那么RabbitMQ就會把這個Message發(fā)送到下一個Consumer。這樣就保證了在Consumer異常退出的情況下數(shù)據(jù)也不會丟失。

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

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

sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged

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

五、注意：

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

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

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

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

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

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

最新評論