需求：
可以對指定機器異步的執(zhí)行多個命令
例子：
>>:run "df -h" --hosts 192.168.3.55 10.4.3.4 task id: 45334 >>: check_task 45334 >>:
注意，每執(zhí)行一條命令，即立刻生成一個任務(wù)ID,不需等待結(jié)果返回，通過命令check_task TASK_ID來得到任務(wù)結(jié)果
README
1 基于RabbitMQ rpc實現(xiàn)的主機管理 2 可以對指定機器異步的執(zhí)行多個命令 3 例子： 4 >>:run "df -h" --hosts 192.168.3.55 10.4.3.4 5 task id: 45334 6 >>: check_task 45334 #查看任務(wù)信息 7 8 程序結(jié)構(gòu): 9 RabbitMQ_PRC/#綜合目錄10 |- - -PRC_CLIENT/#client程序主目錄11 | |- - -__init__.py12 | |- - -bin/#執(zhí)行程目錄13 | | |- - -__init__.py14 | | |- - -clien_start.py #客戶端執(zhí)行文件15 | |16 | |17 | |- - -core #主邏輯程序目錄18 | | |- - -__init__.py19 | | |- - -clien_class.py#客戶端執(zhí)行主要邏輯 類20 | |21 | |22 |23 |24 |- - -PRC_SERVER/#服務(wù)端程序目錄25 | |- - -__init__.py26 | |- - -bin/#執(zhí)行目錄27 | | |- - -__init__.py28 | | |- - -server_start.py#服務(wù)端程序執(zhí)行文件29 | |30 | |31 | |- - -core/##主邏輯程序目錄32 | | |- - -server_class.py#主邏輯 相關(guān)類33 | |34 |35 |- - -README
程序結(jié)構(gòu): RabbitMQ_PRC/#綜合目錄 |- - -PRC_CLIENT/#client程序主目錄 | |- - -__init__.py | |- - -bin/#執(zhí)行程目錄 | | |- - -__init__.py | | |- - -clien_start.py #客戶端執(zhí)行文件
1 import os ,sys2 BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))#獲取相對路徑轉(zhuǎn)為絕對路徑賦于變量3 sys.path.append(BASE_DIR)#增加環(huán)境變量4 5 from core.client_class import Threa6 7 if __name__ == '__main__':8 RPCS=Threa()9 response=RPCS.th_start()
| |- - -core #主邏輯程序目錄 | | |- - -__init__.py | | |- - -clien_class.py#客戶端執(zhí)行主要邏輯 類
1 import pika 2 import uuid 3 import threading 4 import random 5 6 class FibonacciRpcClient(object): 7 def __init__(self): 8 #self.credentials=pika.PlainCredentials("test","test") 9 self.connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))#生成連接的服務(wù)端 ip 10 #self.connection = pika.BlockingConnection(pika.ConnectionParameters("192.168.11.51",15672,'/',self.credentials))#生成連接的服務(wù)端 ip 11 self.channel = self.connection.channel()#創(chuàng)建一個管道 12 13 def get_respon(self,cal_queue,cal_id):#取任務(wù)信息 14 self.response=None 15 self.callback_id=cal_id#隊列名 16 self.channel.basic_consume(self.on_response,queue=cal_queue)# 使用回調(diào)函數(shù) 17 while self.response is None: 18 self.connection.process_data_events()#非阻塞模式接收消息 19 return self.response#返回 20 21 def on_response(self, ch, method, props, body):#回調(diào)函數(shù) 22 if self.callback_id == props.correlation_id:#判斷服務(wù)端返回的隊列名是否與當前所生成的隊列名一致 23 self.response = body# 將服務(wù)端的結(jié)果賦于返回來的結(jié)果變量 24 ch.basic_ack(delivery_tag = method.delivery_tag)##確保消息被 接收 25 26 def call(self, queues,n):#發(fā)送消息的函數(shù) 27 result = self.channel.queue_declare(exclusive=False)#隨機生成一個隊列，收消息后不刪除 28 self.callback_queue = result.method.queue#賦于管道 變量 29 self.corr_id = str(uuid.uuid4())#生成一個服務(wù)端返回消息的隊列名 30 self.channel.basic_publish(exchange='', 31 routing_key=queues,#隊列名 32 properties=pika.BasicProperties( 33 reply_to = self.callback_queue,#發(fā)送的管道隊列名 34 correlation_id = self.corr_id,#發(fā)送給服務(wù)端，用于返回消息的隊列名 35 ), 36 body=str(n))#發(fā)送的內(nèi)容數(shù)據(jù) 37 return self.callback_queue,self.corr_id#返回管道名 隊列id號 38 39 class Threa(object):#線程 類 40 def __init__(self): 41 self.info={}#生成一個字典 42 self.help_info=''' 指令示例