解決RabbitMQ訊息丟失與重複消費問題

最近使用者反饋提交的sql查詢一直處於長時間等待狀態，經過排查觀察，發現部分查詢請求丟失，導致使用者提交的查詢未被正常接收，繼而長時間無響應。

現象：即使sql控制台提交10個簡單sql查詢 -> 訊息傳送方：傳送10條訊息至訊息佇列 -> 訊息消費方：只消費了7條訊息

訊息佇列：rabbitmq：

消費者：python：

結論：訊息傳送正常

排查步驟：檢視log

結論：訊息數量正常

診斷步驟：

執行機安裝rabbitmq-dump-queue外掛程式，用於dump佇列的訊息；

1. 執行機：停止服務；

2. 使用者：提交10個sql查詢：

3. 傳送方：檢視web服務端的輸出日誌，確定10個訊息已經往訊息佇列寫；

4. 執行機：通過rabbitmq-dump-queue檢視佇列的訊息，確認是正常10個訊息寫入；

watch -n 1 '$gopath/src/rabbitmq-dump-queue/rabbitmq-dump-queue -uri="amqp://guest:guest@***xx:5672" -queue ph_open_task'

5. 執行機：啟動服務，訊息佇列中的訊息全部被接收；**邏輯：

try:
pool = pool(processes=40)
def callback(ch, method, properties, body):
try:
dosomething...
except exception as e:
print traceback.format_exc()
logger_msg.info(traceback.format_exc())
finally:
// 這裡會有問題，即使訊息未被處理也會反饋ack給rabbitmq
ch.basic_ack(delivery_tag=method.delivery_tag)
while
true:
try:
connection = pika.blockingconnection(
pika.connectionparameters(host='******xx'))
channel = connection.channel()
channel.queue_declare(queue=queue_name, durable=true)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback, queue=queue_name, no_ack=false)
channel.start_consuming()
except pika.exceptions.connectionclosed as e:
continue
except exception as e:
logger_msg.info(traceback.format_exc())
finally:
channel.basic_ack(delivery_tag=method.delivery_tag)
pool.close
()pool.join
()

結論：本例中消費者主程序將持續監聽mq，一旦mq有訊息將會拉取，隨後從程序池中啟動子程序來處理訊息，但是從程序池啟動子程序的過程並不一定成功（若當前程序池沒有空閒子程序），而主程序不管任何情況下都給mq傳送ack狀態碼，從而mq將未處理的訊息移除掉，導致訊息丟失

問題是在消費者環節產生，因此對消費者做改動，需要調整消費者的架構：

目前方案的問題以及解決方案：

解決RabbitMQ訊息丟失與重複消費問題

RabbitMQ防止訊息丟失

RabbitMQ防止訊息丟失

rabbitmq 重複ACK導致訊息丟失

解決RabbitMQ訊息丟失與重複消費問題

RabbitMQ防止訊息丟失

RabbitMQ防止訊息丟失

rabbitmq 重複ACK導致訊息丟失

相關推薦