python使用pika庫調(diào)用rabbitmq參數(shù)使用詳情

更新時間：2022年08月31日 15:47:24 作者：IT之一小佬

這篇文章主要介紹了python使用pika庫調(diào)用rabbitmq參數(shù)使用詳情，文章通過展開文章主題分享了三種方式，具有一定的參考價值，需要的朋友可以參考一下

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

auto_ack=False
ch.basic_ack(delivery_tag=method.delivery_tag)

生產(chǎn)者模式：

示例代碼：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.創(chuàng)建隊列
channel.queue_declare(queue='hello')
 
# 3.向指定隊列插入數(shù)據(jù)
channel.basic_publish(exchange='',  # 簡單模式
                      routing_key='hello',  # 指定隊列
                      body='Hello World!')  # 向隊列中添加的數(shù)據(jù)
 
print(" [x] Sent 'Hello World!'")

運行結(jié)果：

消費者模式：

示例代碼：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.創(chuàng)建隊列
# 注意：這一步不是必須的，但是如果消費者先啟動而不是生成者先啟動時，這時隊列中還沒有hello隊列，這時就會報錯
channel.queue_declare(queue='hello')
 
# 3.確定回調(diào)函數(shù)
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
# 4.確定監(jiān)聽隊列參數(shù)
channel.basic_consume(queue='hello',
                      auto_ack=False,  # 手動應(yīng)答方式
                      on_message_callback=callback)
 
print(' [*] Waiting for messages. To exit press CTRL+C')
# 5.正式監(jiān)聽
channel.start_consuming()

運行結(jié)果：

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

2、持久化參數(shù)

#聲明queue
channel.queue_declare(queue='hello2', durable=True)  # 若聲明過，則換一個名字
 
channel.basic_publish(exchange='',
                      routing_key='hello2',
                      body='Hello World!',
                      properties=pika.BasicProperties(
                          delivery_mode=2,  # make message persistent
                          )
                      )

生成者方式：

示例代碼：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.創(chuàng)建持久化隊列
# 注意：非持久化隊列不能變持久化隊列，反之也是這樣的，所有創(chuàng)建隊列中不能創(chuàng)建和非持久化隊列重名的隊列
channel.queue_declare(queue='hello2', durable=True)
 
# 3.向指定隊列插入數(shù)據(jù)
channel.basic_publish(exchange='',  # 簡單模式
                      routing_key='hello2',  # 指定隊列
                      body='Hello World!',  # 向隊列中添加的數(shù)據(jù)
                      properties=pika.BasicProperties(
                          delivery_mode=2,  # make message persistent
                      )
                      )
print(" [x] Sent 'Hello World!'")

運行結(jié)果：

消費者方式：

示例代碼：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.創(chuàng)建持久化隊列
# 注意：非持久化隊列不能變持久化隊列，反之也是這樣的，所有創(chuàng)建隊列中不能創(chuàng)建和非持久化隊列重名的隊列
# 注意：這一步不是必須的，但是如果消費者先啟動而不是生成者先啟動時，這時隊列中還沒有hello2隊列，這時就會報錯
channel.queue_declare(queue='hello2', durable=True)

# 3.確定回調(diào)函數(shù)
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)
# 4.確定監(jiān)聽隊列參數(shù)
channel.basic_consume(queue='hello2',  # 指定隊列
                      auto_ack=False,  # 手動應(yīng)答方式
                      on_message_callback=callback)
 
print(' [*] Waiting for messages. To exit press CTRL+C')
# 5.正式監(jiān)聽
channel.start_consuming()

運行結(jié)果：

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

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

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

channel.basic_qos(prefetch_count=1)

生產(chǎn)者模式：

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

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.創(chuàng)建隊列
channel.queue_declare(queue='hello3')
 
# 3.向指定隊列插入數(shù)據(jù)
channel.basic_publish(exchange='',  # 簡單模式
                      routing_key='hello3',  # 指定隊列
                      body='Hello World666!',  # 向隊列中添加的數(shù)據(jù)
                      properties=pika.BasicProperties(
                          delivery_mode=2,  # make message persistent
                      )
                      )
print(" [x] Sent 'Hello World!'")

運行結(jié)果：

消費者模式：

示例代碼1：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.創(chuàng)建隊列
# 注意：這一步不是必須的，但是如果消費者先啟動而不是生成者先啟動時，這時隊列中還沒有hello2隊列，這時就會報錯
channel.queue_declare(queue='hello3')
 
# 3.確定回調(diào)函數(shù)
def callback(ch, method, properties, body):
    import time
    time.sleep(15)
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
# 公平分發(fā)，若不加下面這行代碼，默認是輪詢分發(fā)
channel.basic_qos(prefetch_count=1)
 
# 4.確定監(jiān)聽隊列參數(shù)
channel.basic_consume(queue='hello3',  # 指定隊列
                      auto_ack=False,  # 手動應(yīng)答方式
                      on_message_callback=callback)
 
print(' [*] Waiting for messages. To exit press CTRL+C')
# 5.正式監(jiān)聽
channel.start_consuming()

運行結(jié)果：

示例代碼2：

import pika
 
# 1.連接rabbit
connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.124.104'))
channel = connection.channel()
 
# 2.創(chuàng)建隊列
# 注意：這一步不是必須的，但是如果消費者先啟動而不是生成者先啟動時，這時隊列中還沒有hello2隊列，這時就會報錯
channel.queue_declare(queue='hello3')
 
# 3.確定回調(diào)函數(shù)
def callback(ch, method, properties, body):
    import time
    time.sleep(3)
    print(" [x] Received %r" % body)
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
# 公平分發(fā)，若不加下面這行代碼，默認是輪詢分發(fā)
channel.basic_qos(prefetch_count=1)
 
# 4.確定監(jiān)聽隊列參數(shù)
channel.basic_consume(queue='hello3',  # 指定隊列
                      auto_ack=False,  # 手動應(yīng)答方式
                      on_message_callback=callback)
 
print(' [*] Waiting for messages. To exit press CTRL+C')
# 5.正式監(jiān)聽
channel.start_consuming()

注意：當(dāng)一個py文件執(zhí)行多次時，會有下面提示：