1.服務端

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
"""
@File:      rabbitmq_server.py
@Date:      2025/6/26 10:42
@Author:    xxx
@Description:
1. RabbitMQ服務端，支持多節點命令執行
2. 作為被控節點運行，可接收定向命令并返回結果
"""import ssl
import pika
import time
import json
import socket
import logging
import subprocess
import configparser# 定義日志模塊
logger = logging.getLogger()
# 設置全局日志級別設為最低（DEBUG）
# # 可選: DEBUG, INFO, WARNING, ERROR, CRITICAL
logger.setLevel(logging.DEBUG)# 1. 文件日志（僅輸出到文件）
file_handler = logging.FileHandler('rabbitmq_server.log', encoding='utf-8')
file_handler.setLevel(logging.DEBUG)  # 文件記錄DEBUG及以上級別
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(lineno)04d - %(message)s', datefmt="%Y-%m-%d %H:%M:%S"))
logger.addHandler(file_handler)# 2. 控制臺日志（僅輸出到控制臺）
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)  # 控制臺僅顯示INFO及以上
console_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(lineno)04d - %(message)s', datefmt="%Y-%m-%d %H:%M:%S"))
logger.addHandler(console_handler)RABBITMQ_HOST_CONF = "/etc/rabbitmq/rabbitmq.conf"class RabbitMQServer:"""RabbitMQ RPC服務器類功能：接收并執行來自客戶端的定向命令"""def __init__(self, node_name=None, mq_user="rabbitmq", mq_password="rabbitmq@123",mq_virtual_host="/", mq_host=None, mq_port=5671,mq_ca="/opt/ssl/ca_certificate.pem"):"""初始化RabbitMQ服務端:param node_name: 節點名稱標識（唯一）:param mq_user: RabbitMQ用戶名:param mq_password: RabbitMQ密碼:param mq_virtual_host: 虛擬主機:param mq_host: RabbitMQ服務器IP:param mq_port: RabbitMQ服務端口:param mq_ca: SSL證書路徑"""# 節點配置self.NODE_NAME = node_name if node_name else socket.gethostname()# 連接配置self.RABBITMQ_USER = mq_userself.RABBITMQ_UNLOCK_CODE = mq_passwordself.RABBITMQ_VIRTUAL_HOST = mq_virtual_host# 如果沒有設置RabbitMQ服務器IP，則連接到配置文件中設置的IP節點self.RABBITMQ_HOST = mq_host if mq_host else self.get_option(RABBITMQ_HOST_CONF, "global", "rabbitmq_host")self.RABBITMQ_PORT = mq_portself.SSL_CA_PATH = mq_ca# 初始化連接self._setup_connection()def get_option(self, file_path, section, option):"""獲取 file_path 配置項值，若配置文件沒有，返回空字符串:param section: section字符串，例如：'global':param option: key值，例如：'manage_nodes':return: 字符串類型數據"""parser = configparser.ConfigParser()parser.read(file_path)if not parser.has_option(section, option):return ""else:return parser.get(section, option)def _get_ssl_options(self):"""配置SSL安全連接選項"""# 創建一個 SSL/TLS 安全通信的上下文對象# 生產環境建議指定協議版本，避免使用不安全的默認值context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)# 在 SSL/TLS 通信中承擔核心安全驗證功能context.load_verify_locations(self.SSL_CA_PATH)# 創建 RabbitMQ 的 SSL 連接配置對象return pika.SSLOptions(context,        # ssl.SSLContext 對象，包含證書、協議版本等SSL配置# TODO 可以根據具體證書，設置該選項"localhost"     # 服務器主機名，用于證書驗證（需匹配證書CN或SAN字段），主機名驗證：強制檢查證書中的 CN(Common Name) 或 SAN(Subject Alternative Name) 是否匹配"localhost")def _setup_connection(self):"""建立RabbitMQ連接并設置隊列"""# 創建 RabbitMQ 連接所需的用戶名/密碼認證對，用于后續建立連接時的身份驗證credentials = pika.PlainCredentials(self.RABBITMQ_USER,         # RabbitMQ 服務認證用戶名（字符串類型）self.RABBITMQ_UNLOCK_CODE   # RabbitMQ 服務認證密碼（字符串類型）)# RabbitMQ 連接參數設置connection_params = pika.ConnectionParameters(# RabbitMQ服務器IP地址（字符串類型）# - 若未配置則自動讀取配置文件中的虛擬IP# - 示例：'192.168.120.19' 或 'rabbitmq.example.com'host=self.RABBITMQ_HOST,# RabbitMQ服務端口（整數類型）# - 默認使用加密端口5671# - 非加密連接通常用5672port=self.RABBITMQ_PORT,# 虛擬主機名稱（字符串類型）# - 默認'/'表示根虛擬主機# - 用于多租戶隔離場景virtual_host=self.RABBITMQ_VIRTUAL_HOST,# 認證憑證對象（pika.PlainCredentials）# - 包含用戶名/密碼信息# - 必須與RabbitMQ配置的用戶權限匹配credentials=credentials,# SSL配置對象（pika.SSLOptions）# - 包含CA證書和主機名驗證配置# - 為空時建立非加密連接ssl_options=self._get_ssl_options(),# 心跳檢測間隔（秒，整數類型），建議設置為300-1200秒（防止NAT超時斷開）# - 600秒=10分鐘檢測一次連接活性# - 0表示禁用心跳（不推薦）heartbeat=600   # 需 ≥ 客戶端配置)# 建立與 RabbitMQ 服務器的同步阻塞式連接# 阻塞式：所有操作（如發送/接收消息）會阻塞當前線程直到完成self.connection = pika.BlockingConnection(connection_params)# 在連接上創建 AMQP 信道self.channel = self.connection.channel()# 聲明節點專用隊列（綁定到該信道）# 如果隊列不存在則自動創建，如果存在則直接綁定self.channel.queue_declare(queue=self.NODE_NAME,   # 隊列名稱（使用節點名作為唯一標識）durable=True            # 隊列持久化標志，需要配合消息的delivery_mode=2使用才能完全持久化)# 控制并發數，值越大并發越高，但資源消耗和復雜度也增加# TODO 集群需要調研配置并發度self.channel.basic_qos(prefetch_count=1)# 將當前信道綁定到指定隊列，開始監聽消息，消費消息（通過該信道）self.channel.basic_consume(queue=self.NODE_NAME,   # 指定消費的隊列名稱（當前節點專屬隊列）on_message_callback=self._execute_command,  # 消息處理回調函數auto_ack=False          # 手動消息確認模式)def _execute_command(self, ch, method, props, body):"""執行接收到的命令并返回結果"""try:# 解析消息內容（JSON格式）message = json.loads(body.decode('utf-8'))command = message.get('command', '')    # 獲取要執行的命令target = message.get('target', '')      # 獲取目標節點標識logger.info(f" [x] 收到({target})命令：{command}")# 校驗目標節點（防止誤處理其他節點的消息）if target != self.NODE_NAME:logger.warning(f" [x] 收到非本節點({self.NODE_NAME})命令，已忽略")ch.basic_ack(delivery_tag=method.delivery_tag)  # 確認消息（防止重新投遞）returnlogger.info(f" [*] 執行命令 【{command}】...")try:# 通過子進程執行命令（同步阻塞）output = subprocess.check_output(command,shell=True,     # 啟用shell解析stderr=subprocess.STDOUT,   # 捕獲標準錯誤timeout=60      # 超時時間（秒）)response = output.decode('utf-8')except subprocess.TimeoutExpired:# 超時異常response = "Error: Command timed out"except subprocess.CalledProcessError as e:# 命令執行失敗response = f"Error: {e.output.decode('utf-8')}"except Exception as e:# 其他系統異常response = f"System Error: {str(e)}"# 返回執行結果ch.basic_publish(exchange='',    # 默認交換器routing_key=props.reply_to, # 回復到客戶端指定的回調隊列properties=pika.BasicProperties(correlation_id=props.correlation_id,    # 關聯ID（匹配請求-響應）delivery_mode=2 # 持久化消息到磁盤，會降低性能（服務重啟不丟失）),body=response.encode('utf-8')   # 響應內容)logger.info(f" [*] 命令執行完成")# 確認消息處理完成ch.basic_ack(delivery_tag=method.delivery_tag)except Exception as e:# 全局異常處理logger.exception(f" [x] 消息處理異常: {str(e)}")# 否定確認（可能重新投遞）ch.basic_nack(delivery_tag=method.delivery_tag)def start(self, max_retries=5, retry_delay=10):"""啟動RabbitMQ服務并持續監聽消息功能：管理服務生命周期，處理連接異常和重試邏輯:param max_retries: 最大重試次數，默認5次:param retry_delay: 重試間隔時間(秒)，默認10秒:return:"""# 當前重試次數計數器retry_count = 0# 主服務循環（持續運行直到主動終止）while True:try:# 打印服務狀態信息logger.info(f" [*] {self.NODE_NAME} 節點服務啟動 (嘗試 {retry_count + 1}/{max_retries})")logger.info(f" [*] 等待隊列 {self.NODE_NAME} 中的請求...")# 檢查并重建連接（如果不存在或已關閉）if not hasattr(self, 'connection') or self.connection.is_closed:self._setup_connection()    # 初始化AMQP連接# 開始消費消息（阻塞調用）self.channel.start_consuming()except pika.exceptions.AMQPConnectionError as e:# RabbitMQ連接異常處理retry_count += 1logger.exception(f"連接失敗: {str(e)}")# 超過最大重試次數則終止服務if retry_count >= max_retries:logger.error(" [x] 達到最大重試次數，終止服務")self.close()break   # 退出循環# 未達上限則延遲重試logger.warning(f" [*] {retry_delay}秒后嘗試重新連接...")time.sleep(retry_delay)except KeyboardInterrupt:# 處理用戶主動終止（Ctrl + C）logger.error("\n [x] 接收到終止信號")self.close()logger.error(" [x] 服務已停止")break   # 退出循環except Exception as e:# 其他未捕獲異常處理logger.exception(f"服務異常: {str(e)}")time.sleep(retry_delay) # 防止異常時CPU空轉def close(self):"""安全關閉RabbitMQ連接功能：清理資源，確保連接被正確關閉:return:"""# 防御式編程：檢查連接存在且未關閉if hasattr(self, 'connection') and not self.connection.is_closed:self.connection.close() # 關閉AMQP連接logger.info(" [x] 連接已安全關閉")if __name__ == '__main__':# 服務啟動入口（自動獲取主機名作為節點名）server = RabbitMQServer()try:server.start()except KeyboardInterrupt:logger.error("\n [x] 接收到終止信號")server.close()logger.error(" [x] 服務已停止")

2.客戶端

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
"""
@File:      rabbitmq_client.py
@Date:      2025/6/26 10:43
@Author:    xxx
@Description:
1. RabbitMQ客戶端類，支持向指定節點發送SSH命令
2. 作為控制端運行，可定向發送命令并接收執行結果
"""import ssl
import pika
import time
import uuid
import json
import socket
import logging
import configparser# 定義日志模塊
logger = logging.getLogger()
# 設置全局日志級別設為最低（DEBUG）
# # 可選: DEBUG, INFO, WARNING, ERROR, CRITICAL
logger.setLevel(logging.DEBUG)# 1. 文件日志（僅輸出到文件）
file_handler = logging.FileHandler('rabbitmq_client.log', encoding='utf-8')
file_handler.setLevel(logging.DEBUG)  # 文件記錄DEBUG及以上級別
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(lineno)04d - %(message)s', datefmt="%Y-%m-%d %H:%M:%S"))
logger.addHandler(file_handler)# 2. 控制臺日志（僅輸出到控制臺）
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)  # 控制臺僅顯示INFO及以上
console_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(lineno)04d - %(message)s', datefmt="%Y-%m-%d %H:%M:%S"))
logger.addHandler(console_handler)RABBITMQ_HOST_CONF = "/etc/rabbitmq/rabbitmq.conf"class RabbitMQClient:"""RabbitMQ RPC客戶端類功能：向指定節點發送命令并獲取執行結果"""def __init__(self, mq_user="rabbitmq", mq_password="rabbitmq@123", mq_virtual_host="/",mq_host=None, mq_port=5671, mq_ca="/opt/ssl/ca_certificate.pem"):"""初始化RabbitMQ客戶端:param mq_user: RabbitMQ用戶名:param mq_password: RabbitMQ密碼:param mq_virtual_host: 虛擬主機:param mq_host: RabbitMQ服務器IP:param mq_port: RabbitMQ服務端口:param mq_ca: SSL證書路徑"""# 連接配置self.RABBITMQ_USER = mq_userself.RABBITMQ_UNLOCK_CODE = mq_passwordself.RABBITMQ_VIRTUAL_HOST = mq_virtual_host# 如果沒有設置RabbitMQ服務器IP，則連接到配置文件中設置的IP節點self.RABBITMQ_HOST = mq_host if mq_host else self.get_option(RABBITMQ_HOST_CONF, "global", "rabbitmq_host")self.RABBITMQ_PORT = mq_portself.SSL_CA_PATH = mq_ca# 響應相關變量self.response = Noneself.corr_id = None# 建立連接logger.info(" [x] 正在建立連接 ...")self._connect()logger.info(" [x] 連接建立成功")def get_option(self, file_path, section, option):"""獲取 file_path 配置項值，若配置文件沒有，返回空字符串:param section: section字符串，例如：'global':param option: key值，例如：'manage_nodes':return: 字符串類型數據"""parser = configparser.ConfigParser()parser.read(file_path)if not parser.has_option(section, option):return ""else:return parser.get(section, option)def _connect(self):"""建立RabbitMQ連接并初始化回調隊列功能：配置安全連接參數、創建通信信道、設置消息回調處理:return:"""# 創建SSL安全上下文，強制使用TLS 1.2協議（禁用不安全的老版本協議）ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)# 加載CA根證書用于驗證服務端證書（防止中間人攻擊）ssl_context.load_verify_locations(self.SSL_CA_PATH)# 將SSL配置封裝為RabbitMQ專用的SSLOptions對象# "localhost"參數要求服務端證書必須包含該主機名（CN或SAN字段）ssl_options = pika.SSLOptions(ssl_context, "localhost")# 創建認證憑證對象（明文用戶名/密碼，實際傳輸時會通過SSL加密）credentials = pika.PlainCredentials(self.RABBITMQ_USER,         # RabbitMQ服務用戶名self.RABBITMQ_UNLOCK_CODE   # RabbitMQ服務密碼)# 配置連接參數connection_params = pika.ConnectionParameters(host=self.RABBITMQ_HOST,    # RabbitMQ服務器地址port=self.RABBITMQ_PORT,    # 默認使用5671加密端口virtual_host=self.RABBITMQ_VIRTUAL_HOST,    # 虛擬主機隔離環境credentials=credentials,    # 用戶名密碼憑證ssl_options=ssl_options,    # SSL安全配置heartbeat=60                # 1分鐘心跳檢測（防連接中斷），更頻繁的心跳檢測（如AWS ELB默認60秒空閑超時）)# 建立阻塞式連接（同步操作，會阻塞直到連接成功或拋出異常）self.connection = pika.BlockingConnection(connection_params)# 創建AMQP信道（單個連接可創建多個信道，減少TCP連接開銷）self.channel = self.connection.channel()# 聲明臨時回調隊列（exclusive=True表示連接關閉時自動刪除隊列）result = self.channel.queue_declare(queue='',       # 空隊列名表示由RabbitMQ自動生成exclusive=True  # 獨占隊列，僅當前連接可用)# 保存自動生成的隊列名稱（用于接收服務端響應）self.callback_queue = result.method.queue# 綁定消息消費回調self.channel.basic_consume(queue=self.callback_queue,  # 監聽回調隊列on_message_callback=self._on_response,  # 響應消息處理函數# TODO 生產環境建議：auto_ack=False  改為手動ACK確保消息可靠處理auto_ack=False   # auto_ack=True 自動確認消息（不推薦生產環境使用）)def _on_response(self, ch, method, props, body):"""RPC模式下的響應消息回調處理函數功能：匹配并接收服務端返回的命令執行結果處理邏輯:1.通過correlation_id匹配對應的請求2.將二進制消息體解碼為字符串3.存儲結果供execute_command方法獲取:param ch: (pika.channel.Channel): 接收到消息的信道對象:param method: (pika.spec.Basic.Deliver): 包含投遞信息（如delivery_tag）:param props: (pika.spec.BasicProperties): 消息屬性（含correlation_id等）:param body: (bytes): 消息體內容（服務端返回的執行結果）:return:"""# 校驗消息關聯ID（確保是本請求的響應）try:if self.corr_id == props.correlation_id:# 解碼服務端返回的消息內容（UTF-8編碼）self.response = body.decode('utf-8')# 注意：此處不需要手動ack，因為消息已在服務端處理時ackexcept UnicodeDecodeError as e:self.response = f"解碼失敗: {str(e)}"def execute_command(self, command, target_node=None, timeout=60):"""向指定RabbitMQ節點發送命令并獲取執行結果（RPC模式）:param command (str): 要執行的shell命令字符串（如"ls -l"）:param target_node (str): 目標節點標識，對應服務端的隊列名- 默認None表示發送到當前主機節點:param timeout (int): 等待響應的超時時間（秒），默認60秒:return str: 命令執行結果文本異常:TimeoutError: 超過指定時間未收到響應時拋出AMQP相關異常: 消息發送失敗時拋出向指定節點執行遠程命令"""# 初始化響應存儲和請求IDself.response = None    # 清空之前的響應self.corr_id = str(uuid.uuid4())    # 生成唯一請求標識# 確定目標節點（如果執行命令沒有指定節點，默認發送到當前主機）if not target_node:target_node = socket.gethostname()# 構建RPC請求消息體（JSON格式）（包含命令和目標節點信息）message = {"command": command,         # 要執行的命令"target": target_node,      # 目標節點標識"timestamp": time.time()    # 請求時間戳}# 發送消息到目標節點的專屬隊列self.channel.basic_publish(exchange='',    # 使用默認直連交換機routing_key=target_node,  # 通過隊列名路由到指定節點properties=pika.BasicProperties(reply_to=self.callback_queue,   # 設置回調隊列名correlation_id=self.corr_id,    # 標記請求ID),# JSON序列化消息body=json.dumps(message).encode('utf-8'))# 等待響應（帶超時機制）start_time = time.time()while self.response is None:# 處理網絡事件（非阻塞）self.connection.process_data_events()# 超時檢查if time.time() - start_time > timeout:raise TimeoutError(f"等待節點 {target_node} 響應超時")# 避免CPU空轉，此處sleep不能太長，影響命令返回時效性time.sleep(0.1)return self.response    # 返回執行結果def close(self):"""安全關閉RabbitMQ連接功能:1. 清理網絡連接資源2. 自動刪除臨時隊列（exclusive隊列）3. 防止資源泄漏:return:"""if self.connection and not self.connection.is_closed:# 關閉連接（會觸發信道關閉）self.connection.close()logger.warning(" [x] 連接已關閉")if __name__ == '__main__':# 使用示例client = RabbitMQClient()try:# 向不同節點發送命令nodes = ["node247", "node248", "node249"]for node in nodes:try:logger.info(f"\n向節點 {node} 執行命令: hostname")logger.info(client.execute_command(command="hostname", target_node=node))except Exception as e:logger.exception(f"節點 {node} 執行失敗: {str(e)}")try:logger.info(f"\n向節點 {node} 執行命令: ls -l /opt/")logger.info(client.execute_command(command="ls -l /opt/", target_node=node))except Exception as e:logger.exception(f"節點 {node} 執行失敗: {str(e)}")try:logger.info(f"\n向節點 {node} 執行命令: date")logger.info(client.execute_command(command="date", target_node=node))except Exception as e:logger.exception(f"節點 {node} 執行失敗: {str(e)}")finally:client.close()

3.調用結果

# 服務端，節點啟動
192.168.120.17 node17
192.168.120.18 node18
192.168.120.19 node19python3 rabbitmq_server.py# 客戶端，節點啟動
192.168.120.17 node17python3 rabbitmq_client.py