第python使用pika庫調(diào)用rabbitmq參數(shù)使用詳情目錄1、應(yīng)答參數(shù)2、持久化參數(shù)3、分發(fā)參數(shù)前言：

python使用pika庫調(diào)用rabbitmq的參數(shù)有三種方式，分別如下所述：

1、應(yīng)答參數(shù)

auto_ack=False

ch.basic_ack(delivery_tag=method.delivery_tag)

生產(chǎn)者模式：

示例代碼：

importpika

#1.連接rabbit

connection=pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))

channel=connection.channel()

#2.創(chuàng)建隊(duì)列

channel.queue_declare(queue='hello')

#3.向指定隊(duì)列插入數(shù)據(jù)

channel.basic_publish(exchange='',#簡單模式

routing_key='hello',#指定隊(duì)列

body='HelloWorld!')#向隊(duì)列中添加的數(shù)據(jù)

print("[x]Sent'HelloWorld!'")

運(yùn)行結(jié)果：

消費(fèi)者模式：

示例代碼：

importpika

#1.連接rabbit

connection=pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))

channel=connection.channel()

#2.創(chuàng)建隊(duì)列

#注意：這一步不是必須的，但是如果消費(fèi)者先啟動(dòng)而不是生成者先啟動(dòng)時(shí)，這時(shí)隊(duì)列中還沒有hello隊(duì)列，這時(shí)就會(huì)報(bào)錯(cuò)

channel.queue_declare(queue='hello')

#3.確定回調(diào)函數(shù)

defcallback(ch,method,properties,body):

print("[x]Received%r"%body)

ch.basic_ack(delivery_tag=method.delivery_tag)

#4.確定監(jiān)聽隊(duì)列參數(shù)

channel.basic_consume(queue='hello',

auto_ack=False,#手動(dòng)應(yīng)答方式

on_message_callback=callback)

print('[*]Waitingformessages.ToexitpressCTRL+C')

#5.正式監(jiān)聽

channel.start_consuming()

運(yùn)行結(jié)果：

注意：添加應(yīng)答參數(shù)的好處是當(dāng)消費(fèi)者處理回調(diào)函數(shù)的時(shí)，萬一程序報(bào)錯(cuò)，此時(shí)數(shù)據(jù)就會(huì)消失的。使用應(yīng)答方式后，消費(fèi)者程序萬一報(bào)錯(cuò)，修改完程序后重新啟動(dòng)程序還是可以繼續(xù)消費(fèi)上一次的數(shù)據(jù)的。使用應(yīng)答參數(shù)后，沒處理完一條數(shù)據(jù)都會(huì)給隊(duì)列一個(gè)反饋消息的，也就是說消費(fèi)完一條消息后隊(duì)列才會(huì)刪除這條消息。這種方式效率會(huì)降低一些，根據(jù)項(xiàng)目中數(shù)據(jù)的重要性可以選擇是否需要這個(gè)參數(shù)。

2、持久化參數(shù)

#聲明queue

channel.queue_declare(queue='hello2',durable=True)#若聲明過，則換一個(gè)名字

channel.basic_publish(exchange='',

routing_key='hello2',

body='HelloWorld!',

properties=pika.BasicProperties(

delivery_mode=2,#makemessagepersistent

)

生成者方式：

示例代碼：

importpika

#1.連接rabbit

connection=pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))

channel=connection.channel()

#2.創(chuàng)建持久化隊(duì)列

#注意：非持久化隊(duì)列不能變持久化隊(duì)列，反之也是這樣的，所有創(chuàng)建隊(duì)列中不能創(chuàng)建和非持久化隊(duì)列重名的隊(duì)列

channel.queue_declare(queue='hello2',durable=True)

#3.向指定隊(duì)列插入數(shù)據(jù)

channel.basic_publish(exchange='',#簡單模式

routing_key='hello2',#指定隊(duì)列

body='HelloWorld!',#向隊(duì)列中添加的數(shù)據(jù)

properties=pika.BasicProperties(

delivery_mode=2,#makemessagepersistent

print("[x]Sent'HelloWorld!'")

運(yùn)行結(jié)果：

消費(fèi)者方式：

示例代碼：

importpika

#1.連接rabbit

connection=pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))

channel=connection.channel()

#2.創(chuàng)建持久化隊(duì)列

#注意：非持久化隊(duì)列不能變持久化隊(duì)列，反之也是這樣的，所有創(chuàng)建隊(duì)列中不能創(chuàng)建和非持久化隊(duì)列重名的隊(duì)列

#注意：這一步不是必須的，但是如果消費(fèi)者先啟動(dòng)而不是生成者先啟動(dòng)時(shí)，這時(shí)隊(duì)列中還沒有hello2隊(duì)列，這時(shí)就會(huì)報(bào)錯(cuò)

channel.queue_declare(queue='hello2',durable=True)

#3.確定回調(diào)函數(shù)

defcallback(ch,method,properties,body):

print("[x]Received%r"%body)

ch.basic_ack(delivery_tag=method.delivery_tag)

#4.確定監(jiān)聽隊(duì)列參數(shù)

channel.basic_consume(queue='hello2',#指定隊(duì)列

auto_ack=False,#手動(dòng)應(yīng)答方式

on_message_callback=callback)

print('[*]Waitingformessages.ToexitpressCTRL+C')

#5.正式監(jiān)聽

channel.start_consuming()

運(yùn)行結(jié)果：

注意：加入持久化參數(shù)的好處，當(dāng)rabbitmq隊(duì)列萬一崩了時(shí)，此時(shí)隊(duì)列中的所有數(shù)據(jù)都會(huì)丟失，rabbitmq隊(duì)列中的數(shù)據(jù)是保存在內(nèi)存中，當(dāng)加入持久化參數(shù)后，數(shù)據(jù)將會(huì)保存在硬盤中，rabbitmq崩了或者重啟不會(huì)丟失數(shù)據(jù)。

3、分發(fā)參數(shù)

有兩個(gè)消費(fèi)者同時(shí)監(jiān)聽一個(gè)的隊(duì)列。其中一個(gè)線程sleep2秒，另一個(gè)消費(fèi)者線程sleep1秒，但是處理的消息是一樣多。這種方式叫輪詢分發(fā)（round-robin）不管誰忙，都不會(huì)多給消息，總是你一個(gè)我一個(gè)。想要做到公平分發(fā)（fairdispatch），必須關(guān)閉自動(dòng)應(yīng)答ack，改成手動(dòng)應(yīng)答。使用basicQos(perfetch=1)限制每次只發(fā)送不超過1條消息到同一個(gè)消費(fèi)者，消費(fèi)者必須手動(dòng)反饋告知隊(duì)列，才會(huì)發(fā)送下一個(gè)。

channel.basic_qos(prefetch_count=1)

生產(chǎn)者模式：

示例代碼：【為了產(chǎn)生多條數(shù)據(jù)，將此程序執(zhí)行多次】

importpika

#1.連接rabbit

connection=pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))

channel=connection.channel()

#2.創(chuàng)建隊(duì)列

channel.queue_declare(queue='hello3')

#3.向指定隊(duì)列插入數(shù)據(jù)

channel.basic_publish(exchange='',#簡單模式

routing_key='hello3',#指定隊(duì)列

body='HelloWorld666!',#向隊(duì)列中添加的數(shù)據(jù)

properties=pika.BasicProperties(

delivery_mode=2,#makemessagepersistent

print("[x]Sent'HelloWorld!'")

運(yùn)行結(jié)果：

消費(fèi)者模式：

示例代碼1：

importpika

#1.連接rabbit

connection=pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))

channel=connection.channel()

#2.創(chuàng)建隊(duì)列

#注意：這一步不是必須的，但是如果消費(fèi)者先啟動(dòng)而不是生成者先啟動(dòng)時(shí)，這時(shí)隊(duì)列中還沒有hello2隊(duì)列，這時(shí)就會(huì)報(bào)錯(cuò)

channel.queue_declare(queue='hello3')

#3.確定回調(diào)函數(shù)

defcallback(ch,method,properties,body):

importtime

time.sleep(15)

print("[x]Received%r"%body)

ch.basic_ack(delivery_tag=method.delivery_tag)

#公平分發(fā)，若不加下面這行代碼，默認(rèn)是輪詢分發(fā)

channel.basic_qos(prefetch_count=1)

#4.確定監(jiān)聽隊(duì)列參數(shù)

channel.basic_consume(queue='hello3',#指定隊(duì)列

auto_ack=False,#手動(dòng)應(yīng)答方式

on_message_callback=callback)

print('[*]Waitingformessages.ToexitpressCTRL+C')

#5.正式監(jiān)聽

channel.start_consuming()

運(yùn)行結(jié)果：

示例代碼2：

importpika

#1.連接rabbit

connection=pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))

channel=connection.channel()

#2.創(chuàng)建隊(duì)列

#注意：這一步不是必須的，但是如果消費(fèi)者先啟動(dòng)而不是生成者先啟動(dòng)時(shí)，這時(shí)隊(duì)列中還沒有hello2隊(duì)列，這時(shí)就會(huì)報(bào)錯(cuò)

channel.queue_declare(queue='hello3')

#3.確定回調(diào)函數(shù)

defcallback(ch,method,properties,body):

importtime

time.sleep(3)

print("[x]Received%r"%body)

ch.basic_ack(delivery_tag=method.delivery_tag)

#公平分發(fā)，若不加下面這行代碼，默認(rèn)是輪詢分發(fā)

channel.basic_qos(prefetch_count=