RabbitMQ實踐學習筆記

RabbitMQ實踐

以下是關于RabbitMQ實踐的整理，涵蓋常見場景和示例代碼（基于Markdown格式）。內容按模塊分類，避免步驟詞匯，直接提供可操作的方法：

基礎連接與隊列聲明

使用Python的pika庫建立連接并聲明隊列：

import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')  # 聲明持久化隊列可添加參數 durable=True

消息發布示例：

channel.basic_publish(exchange='',routing_key='hello',body='Hello World!',properties=pika.BasicProperties(delivery_mode=2))  # 消息持久化

工作隊列模式

消費者端的公平分發（prefetch）設置：

channel.basic_qos(prefetch_count=1)  # 每次只處理一條消息
channel.basic_consume(queue='task_queue', on_message_callback=callback)

消息確認機制：

def callback(ch, method, properties, body):print("Received %r" % body)ch.basic_ack(delivery_tag=method.delivery_tag)  # 手動確認

發布/訂閱模式

聲明扇形交換機：

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

臨時隊列綁定：

result = channel.queue_declare(queue='', exclusive=True)
channel.queue_bind(exchange='logs', queue=result.method.queue)

路由與主題模式

直連交換機實現路由鍵過濾：

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
channel.queue_bind(exchange='direct_logs', queue=queue_name, routing_key='error')

主題交換機匹配模式：

channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
channel.queue_bind(exchange='topic_logs', queue=queue_name, routing_key='*.critical')

RPC遠程調用

客戶端發送請求并監聽回調隊列：

result = channel.queue_declare(queue='', exclusive=True)
callback_queue = result.method.queuechannel.basic_publish(exchange='',routing_key='rpc_queue',properties=pika.BasicProperties(reply_to=callback_queue,correlation_id=corr_id,),body=request_body)

服務端響應處理：

def on_request(ch, method, props, body):response = process_request(body)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)

消息TTL與死信隊列

設置消息過期時間：

channel.basic_publish(exchange='',routing_key='queue_with_ttl',body='expiring message',properties=pika.BasicProperties(expiration='60000'))  # 60秒后過期

配置死信交換機：

args = {"x-dead-letter-exchange": "dlx_exchange"}
channel.queue_declare(queue='work_queue', arguments=args)

插件與擴展

延遲消息通過插件實現：

headers = {"x-delay": 5000}  # 延遲5秒
channel.basic_publish(exchange='delayed_exchange',routing_key='delayed_queue',body='delayed msg',properties=pika.BasicProperties(headers=headers))

集群與高可用

配置鏡像隊列策略：

rabbitmqctl set_policy ha-all "^ha\." '{"ha-mode":"all"}'  # 同步所有節點

客戶端連接多個節點：

hosts = [{'host': 'node1'}, {'host': 'node2'}]
connection = pika.BlockingConnection(pika.ConnectionParameters(hosts))

監控與管理

通過API獲取隊列狀態：

import requests
response = requests.get('http://localhost:15672/api/queues', auth=('guest', 'guest'))

以上示例覆蓋了RabbitMQ的核心功能場景，包括消息模式、可靠性保證和擴展功能。實際應用時需根據業務需求調整參數和錯誤處理機制。更多細節可參考RabbitMQ官方文檔和社區最佳實踐。

RabbitMQ 源碼分析（C++ 客戶端）

RabbitMQ 是一個開源的消息代理軟件，采用 Erlang 語言編寫。C++ 客戶端是 RabbitMQ 官方提供的客戶端庫之一，用于在 C++ 項目中與 RabbitMQ 服務器進行交互。以下是對 RabbitMQ C++ 客戶端源碼的分析框架和關鍵點。

源碼結構與核心模塊

RabbitMQ C++ 客戶端的源碼主要包含以下幾個核心模塊：

AMQP 協議實現
C++ 客戶端基于 AMQP 0-9-1 協議實現，核心代碼位于 amqp.h 和 amqp.c 中。這部分負責協議的編解碼、幀的構造與解析。

Socket 通信層
使用系統套接字（Socket）實現與 RabbitMQ 服務器的 TCP 通信。代碼位于 amqp_socket.h 和 amqp_socket.c，支持普通 Socket 和 SSL/TLS 加密通信。

連接管理
amqp_connection.h 中定義了連接的生命周期管理，包括連接建立、心跳檢測、連接關閉等邏輯。

通道管理
通過 amqp_channel.h 實現多路復用機制，單個 TCP 連接可以支持多個邏輯通道（Channel），每個通道獨立處理消息。

消息發布與消費
amqp_queue.h 和 amqp_exchange.h 實現了隊列和交換器的聲明、綁定操作。消息的發布（basic_publish）和消費（basic_consume）邏輯在 amqp_basic.h 中定義。

關鍵流程分析

連接建立流程

調用 amqp_new_connection 創建連接對象。
通過 amqp_socket_open 建立 Socket 連接。
發送協議頭并協商參數（amqp_login）。

消息發布流程

聲明交換器（amqp_exchange_declare）。
構造消息屬性（amqp_basic_properties_t）。
調用 amqp_basic_publish 發送消息幀。

消息消費流程

聲明隊列并綁定到交換器（amqp_queue_bind）。
調用 amqp_basic_consume 訂閱隊列。
通過 amqp_consume_message 循環接收消息。

核心數據結構

幀（Frame）結構
RabbitMQ 使用幀作為通信的基本單位，幀類型包括：

METHOD：傳遞 AMQP 方法（如 queue.declare）。
HEADER：消息屬性。
BODY：消息內容。

連接狀態機
通過 amqp_connection_state_t 管理連接狀態，包括協議版本、通道列表、心跳超時等。

示例代碼片段

以下是使用 C++ 客戶端發布消息的簡化代碼：

#include <amqp.h>
#include <amqp_tcp_socket.h>void publish_message() {amqp_connection_state_t conn = amqp_new_connection();amqp_socket_t *socket = amqp_tcp_socket_new(conn);amqp_socket_open(socket, "localhost", 5672);amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest");amqp_channel_open(conn, 1);amqp_basic_publish(conn, 1, amqp_cstring_bytes("exchange_name"),amqp_cstring_bytes("routing_key"), 0, 0, NULL,amqp_cstring_bytes("Hello, RabbitMQ!"));amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);amqp_connection_close(conn, AMQP_REPLY_SUCCESS);amqp_destroy_connection(conn);
}

性能優化與調試

連接復用
通過通道復用 TCP 連接，減少頻繁建立連接的開銷。

異步模式
使用 amqp_consume_message 的非阻塞模式，結合事件循環實現高吞吐量。

調試工具

啟用 AMQP_DEBUG 宏打印協議幀日志。
使用 Wireshark 抓包分析 AMQP 流量。

參考資源

RabbitMQ 官方文檔
AMQP 0-9-1 協議規范
rabbitmq-c 源碼倉庫

通過分析源碼，可以更深入地理解 RabbitMQ 的通信機制和設計思想，便于定制化開發或性能調優。

基于Python和RabbitMQ的基礎連接與隊列

以下是基于Python和RabbitMQ的基礎連接與隊列聲明的實例代碼示例，涵蓋不同場景和參數配置：

基礎連接與簡單隊列聲明

import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
connection.close()

聲明持久化隊列

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

聲明帶自定義參數的隊列

channel.queue_declare(queue='custom_queue', arguments={'x-message-ttl': 60000})

聲明排他隊列

channel.queue_declare(queue='exclusive_queue', exclusive=True)

聲明自動刪除隊列

channel.queue_declare(queue='auto_delete_queue', auto_delete=True)

聲明隊列并綁定死信交換器

args = {'x-dead-letter-exchange': 'dlx_exchange'}
channel.queue_declare(queue='with_dlx', arguments=args)

聲明優先級隊列

channel.queue_declare(queue='priority_queue', arguments={'x-max-priority': 10})

聲明延遲隊列（通過插件）

channel.queue_declare(queue='delayed_queue', arguments={'x-delayed-type': 'direct'})

聲明長度限制隊列

channel.queue_declare(queue='length_limited', arguments={'x-max-length': 1000})

聲明多消費者隊列

channel.queue_declare(queue='multi_consumer', durable=True)
channel.basic_qos(prefetch_count=1)  # 公平分發

使用SSL加密連接

ssl_params = pika.SSLOptions(ssl.SSLContext(), 'localhost')
connection = pika.BlockingConnection(pika.ConnectionParameters(port=5671, ssl_options=ssl_params))

連接集群節點

params = pika.ConnectionParameters(host='node1.cluster')
connection = pika.BlockingConnection(params)

使用連接池

from rabbitmq import ConnectionPool
pool = ConnectionPool(pika.ConnectionParameters, max_size=10)

聲明隊列并返回隊列信息

result = channel.queue_declare(queue='passive_queue', passive=True)
print(f"Queue has {result.method.message_count} messages")

聲明無參數默認隊列

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

使用異步連接適配器

import pika
connection = pika.SelectConnection(pika.ConnectionParameters('localhost'))

連接失敗重試機制

parameters = pika.ConnectionParameters(host='localhost',connection_attempts=5,retry_delay=3
)

聲明隊列并綁定多個路由鍵

channel.queue_declare(queue='multi_binding')
channel.queue_bind(exchange='logs', queue='multi_binding', routing_key='info')
channel.queue_bind(exchange='logs', queue='multi_binding', routing_key='error')

使用URL參數連接

url = 'amqp://user:pass@host:port/vhost'
parameters = pika.URLParameters(url)
connection = pika.BlockingConnection(parameters)

聲明隊列時設置TTL

args = {'x-expires': 3600000}  # 1小時后自動刪除
channel.queue_declare(queue='temp_queue', arguments=args)

注意事項

所有示例需預先安裝pika庫：pip install pika
生產環境建議添加異常處理邏輯
連接參數應根據實際RabbitMQ服務器配置調整
隊列屬性（如durable）需與已存在隊列的屬性一致，否則會報錯

每個示例均可獨立運行，建議根據實際需求組合使用不同參數。

RabbitMQ 工作隊列模式與 prefetch 設置

RabbitMQ 的工作隊列模式（Work Queue）用于在多個消費者之間分發任務。通過 prefetch 參數可以控制消費者未確認消息的最大數量，從而優化任務分配效率。

Python 實現工作隊列模式

安裝 RabbitMQ 的 Python 客戶端庫：

pip install pika

生產者代碼示例（發送任務）：

import pikaconnection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()channel.queue_declare(queue='task_queue', durable=True)for i in range(10):message = f"Task {i}"channel.basic_publish(exchange='',routing_key='task_queue',body=message,properties=pika.BasicProperties(delivery_mode=2)  # 消息持久化)print(f" [x] Sent {message}")connection.close()

消費者代碼示例（處理任務）：

import pika
import timedef callback(ch, method, properties, body):print(f" [x] Received {body.decode()}")time.sleep(body.count(b'.'))  # 模擬耗時任務print(" [x] Done")ch.basic_ack(delivery_tag=method.delivery_tag)  # 手動確認connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()channel.queue_declare(queue='task_queue', durable=True)
channel.basic_qos(prefetch_count=1)  # 關鍵 prefetch 設置
channel.basic_consume(queue='task_queue', on_message_callback=callback)print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()