Python操作消息隊(duì)列RabbitMQ的完整指南

更新時(shí)間：2026年05月13日 08:43:18 作者：IT策士

在現(xiàn)代分布式系統(tǒng)中,消息隊(duì)列如同人體的神經(jīng)系統(tǒng),負(fù)責(zé)在各個(gè)服務(wù)之間可靠,高效地傳遞信息,本文介紹了使用Python操作RabbitMQ消息隊(duì)列的基礎(chǔ)概念和高級(jí)應(yīng)用場(chǎng)景,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以了解下

在現(xiàn)代分布式系統(tǒng)中，消息隊(duì)列如同人體的神經(jīng)系統(tǒng)，負(fù)責(zé)在各個(gè)服務(wù)之間可靠、高效地傳遞信息。RabbitMQ 作為 AMQP（高級(jí)消息隊(duì)列協(xié)議）的標(biāo)桿實(shí)現(xiàn)，憑借其靈活的路由、可靠的消息投遞和豐富的生態(tài)，成為了架構(gòu)師工具箱中的必備品。本文作為 Python 中間件系列 的開(kāi)篇，將帶你從零開(kāi)始，用 Python 操作 RabbitMQ，從 “Hello World” 直連模式講到延遲隊(duì)列、RPC 等高級(jí)場(chǎng)景，所有示例均附帶可直接運(yùn)行的代碼和詳盡的控制臺(tái)輸出解析。

1. 核心概念速覽

在敲代碼之前，先在大腦中建立一張地圖。RabbitMQ 的核心由三個(gè)角色和兩個(gè)關(guān)鍵組件構(gòu)成：

生產(chǎn)者 (Producer)：發(fā)送消息的應(yīng)用程序。
消費(fèi)者 (Consumer)：接收并處理消息的應(yīng)用程序。
隊(duì)列 (Queue)：RabbitMQ 內(nèi)部的緩沖區(qū)，用于存儲(chǔ)消息。消息一旦進(jìn)入隊(duì)列，就由 RabbitMQ 負(fù)責(zé)保管，直到消費(fèi)者取走。
交換機(jī) (Exchange)：生產(chǎn)者不會(huì)直接把消息發(fā)到隊(duì)列，而是發(fā)給交換機(jī)。交換機(jī)根據(jù)規(guī)則，決定消息路由到一個(gè)或多個(gè)隊(duì)列。主要有四種類(lèi)型：direct、fanout、topic、headers。
綁定 (Binding)：隊(duì)列和交換機(jī)之間的“連線”，在綁定時(shí)會(huì)指定路由鍵 (Routing Key)。交換機(jī)根據(jù)路由鍵和自身類(lèi)型，決定把消息投遞到哪些綁定的隊(duì)列。

一句話總結(jié)：生產(chǎn)者 -> 交換機(jī) -(通過(guò)綁定和路由鍵)-> 隊(duì)列 <- 消費(fèi)者

理解了這張圖，后面的代碼就只是把它翻譯成 Python 語(yǔ)言而已。

2. 環(huán)境準(zhǔn)備

我們使用 Docker 快速啟動(dòng) RabbitMQ，并安裝 Python 客戶端 pika。

啟動(dòng) RabbitMQ（帶管理界面的版本）：

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

啟動(dòng)后，可通過(guò) http://localhost:15672 訪問(wèn)管理界面，默認(rèn)賬號(hào)密碼 guest/guest。

安裝 pika 庫(kù)：

接下來(lái)，所有代碼示例都將圍繞這兩個(gè)環(huán)境展開(kāi)。

3. 第一章：Hello World —— 最簡(jiǎn)單的消息模型

我們先從一個(gè)最簡(jiǎn)單的 單生產(chǎn)者 -> 單消費(fèi)者 模型開(kāi)始，發(fā)送一句 “Hello World”。

生產(chǎn)者send.py

import pika

# 1. 建立連接
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 2. 聲明隊(duì)列（如果隊(duì)列不存在則創(chuàng)建，存在則復(fù)用）
channel.queue_declare(queue='hello')

# 3. 發(fā)布消息
channel.basic_publish(exchange='',      # 使用默認(rèn)交換機(jī)
                      routing_key='hello', # 消息直接發(fā)到 'hello' 隊(duì)列
                      body='Hello World!')

print(" [x] Sent 'Hello World!'")

# 4. 關(guān)閉連接
connection.close()

消費(fèi)者receive.py

import pika

# 1. 建立連接
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 2. 聲明隊(duì)列（冪等操作，確保隊(duì)列存在）
channel.queue_declare(queue='hello')

# 3. 定義回調(diào)函數(shù)
def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")

# 4. 訂閱隊(duì)列
channel.basic_consume(queue='hello',
                      auto_ack=True,  # 自動(dòng)確認(rèn)（先這樣，后面會(huì)講）
                      on_message_callback=callback)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

運(yùn)行與輸出解析

打開(kāi)兩個(gè)終端：

終端 1（消費(fèi)者）：

$ python receive.py
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received Hello World!

終端 2（生產(chǎn)者）：

$ python send.py
 [x] Sent 'Hello World!'

這里的關(guān)鍵點(diǎn)：

生產(chǎn)者使用默認(rèn)交換機(jī)（空字符串 ''），此時(shí) routing_key 就是目標(biāo)隊(duì)列名。
消費(fèi)者用 basic_consume 持續(xù)監(jiān)聽(tīng)隊(duì)列，auto_ack=True 表示消息一旦送出就自動(dòng)確認(rèn)刪除。

4. 第二章：工作隊(duì)列 —— 任務(wù)的分發(fā)與負(fù)載均衡

真實(shí)世界中，單個(gè)消費(fèi)者往往忙不過(guò)來(lái)。工作隊(duì)列允許多個(gè)消費(fèi)者從同一個(gè)隊(duì)列中取任務(wù)，消息只被一個(gè)消費(fèi)者處理。

循環(huán)分發(fā)（Round-robin）

默認(rèn)情況下，RabbitMQ 會(huì)把消息輪流發(fā)給所有消費(fèi)者。我們創(chuàng)建持續(xù)發(fā)送任務(wù)的 new_task.py 和多個(gè) worker.py。

生產(chǎn)者 new_task.py：

import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue')

for i in range(1, 6):
    message = f"Task {i}"
    channel.basic_publish(exchange='', routing_key='task_queue', body=message)
    print(f" [x] Sent '{message}'")
    time.sleep(0.5)

connection.close()

消費(fèi)者 worker.py：

import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue')

def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")
    # 模擬處理耗時(shí)（偶數(shù)任務(wù)耗時(shí)短，奇數(shù)耗時(shí)長(zhǎng)）
    time.sleep(2 if int(body.decode().split()[1]) % 2 != 0 else 0.5)
    print(f" [x] Done {body.decode()}")
    ch.basic_ack(delivery_tag=method.delivery_tag) # 手動(dòng)確認(rèn)

# 重要：每次只分發(fā)一條消息，消費(fèi)者處理完并確認(rèn)后才發(fā)下一條
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=callback)

print(' [*] Waiting for messages...')
channel.start_consuming()

運(yùn)行與公平分發(fā)演示

打開(kāi)三個(gè)終端：一個(gè)生產(chǎn)，兩個(gè)消費(fèi)。

終端 1（Worker A）：

$ python worker.py
 [*] Waiting for messages...
 [x] Received Task 1     # 耗時(shí)2秒
 [x] Done Task 1
 [x] Received Task 3     # 耗時(shí)2秒
 [x] Done Task 3
 [x] Received Task 5     # 耗時(shí)2秒

終端 2（Worker B）：

$ python worker.py
 [*] Waiting for messages...
 [x] Received Task 2     # 耗時(shí)0.5秒，先完成
 [x] Done Task 2
 [x] Received Task 4     # 耗時(shí)0.5秒
 [x] Done Task 4

終端 3（生產(chǎn)者）：

$ python new_task.py
 [x] Sent 'Task 1'
 [x] Sent 'Task 2'
 [x] Sent 'Task 3'
 [x] Sent 'Task 4'
 [x] Sent 'Task 5'

輸出解讀：

我們沒(méi)有使用自動(dòng)確認(rèn) auto_ack，而是手動(dòng) basic_ack。這保證了如果 Worker A 在處理 Task 1 時(shí)崩潰，該消息會(huì)重新入隊(duì)并分發(fā)給 Worker B。
basic_qos(prefetch_count=1) 是關(guān)鍵。它告訴 RabbitMQ：不要同時(shí)給我超過(guò) 1 條消息。這樣 Worker A 在處理 Task 1 時(shí)，盡管 Task 2、3 已經(jīng)入隊(duì)，但 Task 3 不會(huì)預(yù)發(fā)給 A，而是會(huì)發(fā)給空閑的 Worker B。這就實(shí)現(xiàn)了公平分發(fā)，處理速度快的消費(fèi)者將拿到更多任務(wù)。

5. 第三章：發(fā)布/訂閱 —— 用 Fanout 交換機(jī)廣播消息

現(xiàn)在，我們需要讓多個(gè)消費(fèi)者都能收到同一條消息，就像日志廣播。這需要引入交換機(jī)。fanout 交換機(jī)將消息廣播到所有綁定的隊(duì)列。

生產(chǎn)者emit_log.py

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 聲明 fanout 交換機(jī)
channel.exchange_declare(exchange='logs', exchange_type='fanout')

message = "info: Hello Fanout!"
channel.basic_publish(exchange='logs', routing_key='', body=message)
print(f" [x] Sent {message}")

connection.close()

消費(fèi)者receive_logs.py

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='logs', exchange_type='fanout')

# 讓 RabbitMQ 生成一個(gè)唯一的、臨時(shí)隊(duì)列，消費(fèi)者斷開(kāi)后自動(dòng)刪除
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue

# 綁定隊(duì)列到交換機(jī)
channel.queue_bind(exchange='logs', queue=queue_name)

print(f' [*] Waiting for logs on queue: {queue_name}')

def callback(ch, method, properties, body):
    print(f" [x] {body.decode()}")

channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.start_consuming()

運(yùn)行演示：一發(fā)多收

同時(shí)啟動(dòng)兩個(gè) receive_logs.py，再執(zhí)行 emit_log.py。

消費(fèi)者終端 1：

$ python receive_logs.py
 [*] Waiting for logs on queue: amq.gen-JzTY20BRgKO-HjmUJj0wLg
 [x] info: Hello Fanout!

消費(fèi)者終端 2：

$ python receive_logs.py
 [*] Waiting for logs on queue: amq.gen-0cHw5VhC7KnpjRzAsj0Xww
 [x] info: Hello Fanout!

生產(chǎn)者終端：

$ python emit_log.py
 [x] Sent info: Hello Fanout!

可見(jiàn)，兩個(gè)消費(fèi)者都收到了相同的消息。關(guān)鍵在于 queue_declare(queue='', exclusive=True)：每個(gè)消費(fèi)者啟動(dòng)時(shí)都會(huì)創(chuàng)建一個(gè)隨機(jī)的獨(dú)占臨時(shí)隊(duì)列，并綁定到 logs 交換機(jī)。生產(chǎn)者完全不需要關(guān)心消費(fèi)者的數(shù)量和地址，達(dá)到了徹底的解耦。

6. 第四章：路由模式 —— 用 Direct 交換機(jī)精準(zhǔn)投遞

Fanout 是“無(wú)腦廣播”，而 direct 交換機(jī)根據(jù)完全匹配的路由鍵，將消息投遞給綁定鍵相同的隊(duì)列。適用于按日志級(jí)別（error、info）分發(fā)。

生產(chǎn)者emit_direct_log.py

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'

channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)

print(f" [x] Sent {severity}:{message}")
connection.close()

消費(fèi)者receive_direct_log.py

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue

# 通過(guò)命令行參數(shù)指定綁定的路由鍵，如 python receive_direct_log.py error info
severities = sys.argv[1:] if len(sys.argv) > 1 else ['info']
for severity in severities:
    channel.queue_bind(exchange='direct_logs', queue=queue_name, routing_key=severity)

print(f' [*] Waiting for {severities} logs. Queue: {queue_name}')

def callback(ch, method, properties, body):
    print(f" [x] {method.routing_key}:{body.decode()}")

channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.start_consuming()

運(yùn)行演示

終端 1（只收 error）：

$ python receive_direct_log.py error
 [*] Waiting for ['error'] logs. Queue: amq.gen-XXX

終端 2（收 error 和 info）：

$ python receive_direct_log.py error info
 [*] Waiting for ['error', 'info'] logs. Queue: amq.gen-YYY

生產(chǎn)者發(fā)送消息：

$ python emit_direct_log.py error "Disk full"
 [x] Sent error:Disk full
$ python emit_direct_log.py info "Server started"
 [x] Sent info:Server started

輸出：終端1只收到 error 消息，終端2則兩條都收到。 這種精確匹配非常適合按模塊、優(yōu)先級(jí)區(qū)分處理。

7. 第五章：主題模式 —— 用 Topic 交換機(jī)實(shí)現(xiàn)靈活匹配

topic 交換機(jī)是 direct 的升級(jí)版。路由鍵是由點(diǎn)分隔的單詞列表（如 “weather.us.east”），綁定鍵可以使用通配符：

* 匹配剛好一個(gè)單詞。
# 匹配零個(gè)或多個(gè)單詞。

生產(chǎn)者emit_topic.py

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')

routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'

channel.basic_publish(exchange='topic_logs', routing_key=routing_key, body=message)
print(f" [x] Sent {routing_key}:{message}")
connection.close()

消費(fèi)者receive_topic.py

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')

result = channel.queue_declare('', exclusive=True)
queue_name = result.method.queue

# 綁定鍵從命令行接收，例如: "kern.*" 或 "*.critical" 或 "#"
binding_keys = sys.argv[1:] if len(sys.argv) > 1 else ['anonymous.*']
for binding_key in binding_keys:
    channel.queue_bind(exchange='topic_logs', queue=queue_name, routing_key=binding_key)

print(f' [*] Waiting for logs. Binding keys: {binding_keys}. Queue: {queue_name}')

def callback(ch, method, properties, body):
    print(f" [x] {method.routing_key}:{body.decode()}")

channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.start_consuming()

運(yùn)行演示

消費(fèi)者 1：訂閱所有 kern 開(kāi)頭的日志

$ python receive_topic.py "kern.*"
 [*] Waiting for logs. Binding: ['kern.*']

消費(fèi)者 2：訂閱所有 critical 結(jié)尾的日志

$ python receive_topic.py "*.critical"
 [*] Waiting for logs. Binding: ['*.critical']

消費(fèi)者 3：接收所有日志（#）

$ python receive_topic.py "#"
 [*] Waiting for logs. Binding: ['#']
**生產(chǎn)者：**
```bash
$ python emit_topic.py kern.critical "Kernel panic"
 [x] Sent kern.critical:Kernel panic
$ python emit_topic.py app.critical "App crash"
 [x] Sent app.critical:App crash
$ python emit_topic.py kern.info "Kernel info"
 [x] Sent kern.info:Kernel info

結(jié)果：

kern.critical 會(huì)被三個(gè)消費(fèi)者都收到（匹配 kern.*, *.critical, #）。
app.critical 僅被消費(fèi)者2和3收到。
kern.info 僅被消費(fèi)者1和3收到。

Topic 交換機(jī)提供了極其強(qiáng)大的基于模式的消息路由，是構(gòu)建事件驅(qū)動(dòng)架構(gòu)的利器。

8. 第六章：消息可靠性 —— 確認(rèn)、持久化與發(fā)布者確認(rèn)

生產(chǎn)環(huán)境中，消息不能丟。RabbitMQ 提供了三重保障：

消費(fèi)者確認(rèn)（Acknowledgement）：消費(fèi)者告訴 RabbitMQ，消息已成功處理。我們已在工作隊(duì)列中使用 basic_ack。
隊(duì)列持久化（Durable）：RabbitMQ 重啟后隊(duì)列不丟失。queue_declare(queue='task_queue', durable=True)。
消息持久化：消息本身標(biāo)記為持久，寫(xiě)入磁盤(pán)。properties=pika.BasicProperties(delivery_mode=2)。
發(fā)布者確認(rèn)（Publisher Confirms）：生產(chǎn)者知道消息是否成功到達(dá) RabbitMQ。

我們來(lái)改造工作隊(duì)列，加入發(fā)布者確認(rèn)和持久化。

可靠生產(chǎn)者reliable_send.py

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 啟用發(fā)布者確認(rèn)
channel.confirm_delivery()

# 聲明持久化隊(duì)列
channel.queue_declare(queue='durable_task', durable=True)

for i in range(1, 4):
    message = f"Persistent Task {i}"
    # 將消息標(biāo)記為持久化
    properties = pika.BasicProperties(delivery_mode=2)
    try:
        channel.basic_publish(exchange='',
                              routing_key='durable_task',
                              body=message,
                              properties=properties)
        print(f" [x] Confirmed: {message}")
    except pika.exceptions.UnroutableError:
        print(f" [x] Failed to route: {message}")

connection.close()

可靠消費(fèi)者reliable_worker.py

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

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

def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")
    # 模擬處理
    import time
    time.sleep(1)
    print(f" [x] Done {body.decode()}")
    # 手動(dòng)確認(rèn)
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='durable_task', on_message_callback=callback)

print(' [*] Waiting for durable tasks...')
channel.start_consuming()

運(yùn)行演示

先啟動(dòng) reliable_worker.py，再執(zhí)行 reliable_send.py：

生產(chǎn)者輸出：

$ python reliable_send.py
 [x] Confirmed: Persistent Task 1
 [x] Confirmed: Persistent Task 2
 [x] Confirmed: Persistent Task 3

現(xiàn)在即使重啟 RabbitMQ（docker restart rabbitmq），隊(duì)列和尚未被消費(fèi)的持久化消息也不會(huì)丟失。

提示：持久化有性能開(kāi)銷(xiāo)，只對(duì)關(guān)鍵消息使用。

9. 第七章：高級(jí)應(yīng)用 —— 死信隊(duì)列與延遲隊(duì)列

如何實(shí)現(xiàn)“訂單30分鐘未支付自動(dòng)取消”這樣的延遲任務(wù)？RabbitMQ 本身沒(méi)有延遲隊(duì)列，但可以用 死信交換機(jī)（DLX） 和 消息TTL（存活時(shí)間） 組合實(shí)現(xiàn)。

原理

給隊(duì)列設(shè)置 x-dead-letter-exchange 和 x-dead-letter-routing-key。當(dāng)消息在該隊(duì)列中變成死信（被拒絕、過(guò)期或隊(duì)列滿）時(shí)，會(huì)被自動(dòng)轉(zhuǎn)發(fā)到死信交換機(jī)。
給消息設(shè)置 expiration (毫秒)。消息過(guò)期后會(huì)變成死信，從而被投遞到指定的延遲隊(duì)列。

延遲隊(duì)列代碼delay_queue.py

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 1. 定義死信交換機(jī)
dlx_exchange = 'dlx_exchange'
channel.exchange_declare(exchange=dlx_exchange, exchange_type='direct')

# 2. 死信隊(duì)列（真正延遲后消費(fèi)的隊(duì)列）
dlx_queue = 'delayed_queue'
channel.queue_declare(queue=dlx_queue)
channel.queue_bind(exchange=dlx_exchange, queue=dlx_queue, routing_key='delayed_key')

# 3. 普通隊(duì)列，設(shè)置死信參數(shù)和隊(duì)列TTL（也可針對(duì)單獨(dú)消息設(shè)TTL）
args = {
    'x-dead-letter-exchange': dlx_exchange,
    'x-dead-letter-routing-key': 'delayed_key',
    # 'x-message-ttl': 5000  # 統(tǒng)一隊(duì)列TTL 5秒，這里用消息TTL演示
}
normal_queue = 'normal_queue'
channel.queue_declare(queue=normal_queue, arguments=args)

# 生產(chǎn)者發(fā)送一條TTL為5秒的消息
message = "Delayed order cancel"
properties = pika.BasicProperties(expiration='5000')  # 消息TTL 5秒
channel.basic_publish(exchange='', routing_key=normal_queue, body=message, properties=properties)
print(f" [x] Sent to normal_queue with 5s TTL: {message}")

# 消費(fèi)者監(jiān)聽(tīng)死信隊(duì)列（即延遲后的隊(duì)列）
def consume_delayed(ch, method, properties, body):
    print(f" [x] Received delayed message at {time.strftime('%X')}: {body.decode()}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

import time
print(f" [*] Start time: {time.strftime('%X')}")
channel.basic_consume(queue=dlx_queue, on_message_callback=consume_delayed)
channel.start_consuming()

運(yùn)行與輸出

$ python delay_queue.py
 [x] Sent to normal_queue with 5s TTL: Delayed order cancel
 [*] Start time: 14:32:10
 [x] Received delayed message at 14:32:15: Delayed order cancel

時(shí)間上正好相差5秒，延遲消費(fèi)達(dá)成！這個(gè)模式廣泛應(yīng)用于定時(shí)觸發(fā)、超時(shí)處理等業(yè)務(wù)。

10. 第八章：RPC —— 遠(yuǎn)程過(guò)程調(diào)用

RPC 允許客戶端將請(qǐng)求消息發(fā)送到隊(duì)列，然后阻塞等待服務(wù)器返回響應(yīng)。實(shí)現(xiàn)的關(guān)鍵是 correlation_id（關(guān)聯(lián)ID）和 reply_to（回調(diào)隊(duì)列）。

服務(wù)器rpc_server.py

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='rpc_queue')

def on_request(ch, method, props, body):
    n = int(body)
    print(f" [.] fib({n})")
    # 模擬計(jì)算斐波那契
    response = fib(n)

    # 將結(jié)果發(fā)回給 props.reply_to 指定的回調(diào)隊(duì)列
    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id=props.correlation_id),
                     body=str(response))
    ch.basic_ack(delivery_tag=method.delivery_tag)

def fib(n):
    if n == 0: return 0
    elif n == 1: return 1
    else: return fib(n-1) + fib(n-2)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)

print(" [x] Awaiting RPC requests")
channel.start_consuming()

客戶端rpc_client.py

import pika
import uuid

class FibonacciRpcClient:
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.channel = self.connection.channel()

        # 創(chuàng)建唯一的回調(diào)隊(duì)列
        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(queue=self.callback_queue,
                                   on_message_callback=self.on_response,
                                   auto_ack=True)

    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body.decode()

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue',
                                   properties=pika.BasicProperties(
                                       reply_to=self.callback_queue,
                                       correlation_id=self.corr_id,
                                   ),
                                   body=str(n))
        # 等待響應(yīng)
        while self.response is None:
            self.connection.process_data_events()
        return int(self.response)

# 使用客戶端
fib_client = FibonacciRpcClient()
print(" [x] Requesting fib(10)")
response = fib_client.call(10)
print(f" [.] Got {response}")

運(yùn)行與輸出

服務(wù)器：

$ python rpc_server.py
 [x] Awaiting RPC requests
 [.] fib(10)

客戶端：

$ python rpc_client.py
 [x] Requesting fib(10)
 [.] Got 55

客戶端發(fā)送請(qǐng)求后，阻塞等待來(lái)自自己專(zhuān)屬回調(diào)隊(duì)列的響應(yīng)，通過(guò) correlation_id 精準(zhǔn)匹配請(qǐng)求與響應(yīng)。這是典型的消息同步模式。

11. 結(jié)語(yǔ)與最佳實(shí)踐

我們由淺入深地走完了 RabbitMQ 在 Python 中六大消息模式與高級(jí)特性。這些代碼片段不僅是示例，更是可以直接復(fù)用到生產(chǎn)項(xiàng)目中的模板。最后，總結(jié)幾條金科玉律：

多用臨時(shí)隊(duì)列，善用交換機(jī)：消費(fèi)者盡量使用隨機(jī)隊(duì)列并綁定到交換機(jī)，實(shí)現(xiàn)組件間解耦。
生產(chǎn)者確認(rèn) + 消費(fèi)者手動(dòng)確認(rèn) + 持久化：三者缺一，高可靠無(wú)從談起。auto_ack 僅在可容忍消息丟失的場(chǎng)景使用。
合理設(shè)置 prefetch_count：避免一個(gè)消費(fèi)者被堆積的消息撐死，是實(shí)現(xiàn)公平調(diào)度的利器。
死信隊(duì)列不僅是延遲任務(wù)：它還適用于收集處理異常的消息，方便排查和人工干預(yù)。
連接與通道管理：BlockingConnection 適合簡(jiǎn)單腳本，異步場(chǎng)景請(qǐng)用 AsyncioConnection 或 TornadoConnection。生產(chǎn)環(huán)境務(wù)必配置心跳和自動(dòng)重連。
監(jiān)控為王：善用 http://localhost:15672 管理界面，觀察隊(duì)列長(zhǎng)度、消息速率、消費(fèi)者數(shù)量，是調(diào)優(yōu)和排錯(cuò)的眼睛。

消息隊(duì)列是分布式系統(tǒng)的脊梁，而 RabbitMQ 這條脊梁足夠強(qiáng)壯且靈活。掌握它，你就掌握了一種構(gòu)建健壯、可擴(kuò)展系統(tǒng)的核心能力。

以上就是Python操作消息隊(duì)列RabbitMQ的完整指南的詳細(xì)內(nèi)容，更多關(guān)于Python操作消息隊(duì)列RabbitMQ的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章:

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

Python操作消息隊(duì)列RabbitMQ的完整指南

目錄

1. 核心概念速覽

2. 環(huán)境準(zhǔn)備

3. 第一章：Hello World —— 最簡(jiǎn)單的消息模型

生產(chǎn)者send.py

消費(fèi)者receive.py

運(yùn)行與輸出解析

4. 第二章：工作隊(duì)列 —— 任務(wù)的分發(fā)與負(fù)載均衡

循環(huán)分發(fā)（Round-robin）

運(yùn)行與公平分發(fā)演示

5. 第三章：發(fā)布/訂閱 —— 用 Fanout 交換機(jī)廣播消息

生產(chǎn)者emit_log.py

消費(fèi)者receive_logs.py

運(yùn)行演示：一發(fā)多收

6. 第四章：路由模式 —— 用 Direct 交換機(jī)精準(zhǔn)投遞

生產(chǎn)者emit_direct_log.py

消費(fèi)者receive_direct_log.py

運(yùn)行演示

7. 第五章：主題模式 —— 用 Topic 交換機(jī)實(shí)現(xiàn)靈活匹配

生產(chǎn)者emit_topic.py

消費(fèi)者receive_topic.py

運(yùn)行演示

8. 第六章：消息可靠性 —— 確認(rèn)、持久化與發(fā)布者確認(rèn)

可靠生產(chǎn)者reliable_send.py

可靠消費(fèi)者reliable_worker.py

運(yùn)行演示

9. 第七章：高級(jí)應(yīng)用 —— 死信隊(duì)列與延遲隊(duì)列

原理

延遲隊(duì)列代碼delay_queue.py

運(yùn)行與輸出

10. 第八章：RPC —— 遠(yuǎn)程過(guò)程調(diào)用

服務(wù)器rpc_server.py

客戶端rpc_client.py

運(yùn)行與輸出

11. 結(jié)語(yǔ)與最佳實(shí)踐

相關(guān)文章

最新評(píng)論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

8. 第六章：消息可靠性 —— 確認(rèn)、持久化與發(fā)布者確認(rèn)