python第六十三天-- 第十一周作業
阿新 • • 發佈：2017-07-05
open 同時 完成 exclusive lba [0 publish clinet color 
題目：基於RabbitMQ rpc實現的主機管理
需求：
可以對指定機器異步的執行多個命令
例子：
>>:run "df -h" --hosts 192.168.3.55 10.4.3.4
task id: 45334
>>: check_task 45334
>>:
註意，每執行一條命令，即立刻生成一個任務ID,不需等待結果返回，通過命令check_task TASK_ID來得到任務結果
README
1 基於RabbitMQ rpc實現的主機管理 2 可以對指定機器異步的執行多個命令 3 例子： 4 >>:run "df -h" --hosts 192.168.3.55 10.4.3.4View Code5 task id: 45334 6 >>: check_task 45334 #查看任務信息 7 8 程序結構: 9 RabbitMQ_PRC/#綜合目錄 10 |- - -PRC_CLIENT/#client程序主目錄 11 | |- - -__init__.py 12 | |- - -bin/#執行程目錄 13 | | |- - -__init__.py 14 | | |- - -clien_start.py #客戶端執行文件 15 | | 16 | | 17 | |- - -core #主邏輯程序目錄18 | | |- - -__init__.py 19 | | |- - -clien_class.py#客戶端執行主要邏輯 類 20 | | 21 | | 22 | 23 | 24 |- - -PRC_SERVER/#服務端程序目錄 25 | |- - -__init__.py 26 | |- - -bin/#執行目錄 27 | | |- - -__init__.py 28 | | |- - -server_start.py#服務端程序執行文件 29 | |30 | | 31 | |- - -core/##主邏輯程序目錄 32 | | |- - -server_class.py#主邏輯 相關類 33 | | 34 | 35 |- - -README
程序結構:
RabbitMQ_PRC/#綜合目錄
|- - -PRC_CLIENT/#client程序主目錄
| |- - -__init__.py
| |- - -bin/#執行程目錄
| | |- - -__init__.py
| | |- - -clien_start.py #客戶端執行文件
1 import os ,sys 2 BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))#獲取相對路徑轉為絕對路徑賦於變量 3 sys.path.append(BASE_DIR)#增加環境變量 4 5 from core.client_class import Threa 6 7 if __name__ == ‘__main__‘: 8 RPCS=Threa() 9 response=RPCS.th_start()View Code
| |- - -core #主邏輯程序目錄
| | |- - -__init__.py
| | |- - -clien_class.py#客戶端執行主要邏輯 類
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"))#生成連接的服務端 ip 10 #self.connection = pika.BlockingConnection(pika.ConnectionParameters("192.168.11.51",15672,‘/‘,self.credentials))#生成連接的服務端 ip 11 self.channel = self.connection.channel()#創建一個管道 12 13 def get_respon(self,cal_queue,cal_id):#取任務信息 14 self.response=None 15 self.callback_id=cal_id#隊列名 16 self.channel.basic_consume(self.on_response,queue=cal_queue)# 使用回調函數 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):#回調函數 22 if self.callback_id == props.correlation_id:#判斷服務端返回的隊列名是否與當前所生成的隊列名一致 23 self.response = body# 將服務端的結果賦於返回來的結果變量 24 ch.basic_ack(delivery_tag = method.delivery_tag)##確保消息被 接收 25 26 def call(self, queues,n):#發送消息的函數 27 result = self.channel.queue_declare(exclusive=False)#隨機生成一個隊列，收消息後不刪除 28 self.callback_queue = result.method.queue#賦於管道 變量 29 self.corr_id = str(uuid.uuid4())#生成一個服務端返回消息的隊列名 30 self.channel.basic_publish(exchange=‘‘, 31 routing_key=queues,#隊列名 32 properties=pika.BasicProperties( 33 reply_to = self.callback_queue,#發送的管道隊列名 34 correlation_id = self.corr_id,#發送給服務端，用於返回消息的隊列名 35 ), 36 body=str(n))#發送的內容數據 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=‘‘‘ 指令示例\033[36;1m 43 run "df -h" --hosts 192.168.3.55 10.4.3.4 44 --- ------- ------- ------------ -------- 45 運行 指令 主機 ip 1# ip 2# 46 check_task_all #查看任務列表 47 check_task 25413 #查看具體id任務信息，過後刪除 48 helps #查看指令幫助 49 \033[0m‘‘‘ 50 51 def check_task_all(self,cmd):#查看所有任務信息 52 53 for i in self.info: 54 print("任務id:%s,服務端:%s,命令:%s"%(i,self.info[i][0],self.info[i][1])) 55 def check_task(self,take_id):#查看任務 56 try: 57 id=int(take_id.split()[1])#取任務ID 58 #print(id,‘任務ID‘) 59 cal_queue=self.info[id][2]#管道名 60 #print(cal_queue,‘隊列‘) 61 cal_id=self.info[id][3]#消息隊列位置 62 #print(cal_id,‘消息位置‘) 63 clinets=FibonacciRpcClient()#調用類 64 rest=clinets.get_respon(cal_queue,cal_id)#取任務信息 65 print(‘任務執行結果:‘,rest.decode())#打印 66 del self.info[id]#從字典中刪除對應任務 67 except Exception as e: 68 print(e) 69 return 70 71 def run(self,str_l):#run函數 72 addr_l=self.attr_l(str_l)#獲取IP 73 oreds=self.oreds_(str_l)#獲取 命令 74 #print(oreds,‘上傳命令‘) 75 for i in addr_l:#取出IP 76 tak_id=random.randint(10000,99999)#任務ID生成 77 #print(tak_id,‘任務ID‘) 78 obj=FibonacciRpcClient()#生成連接類 79 r=obj.call(i,oreds)#ip做隊列名 命令 80 self.info[tak_id]=[i,oreds,r[0],r[1]]#寫入字典 tak_id{ ip 命令 管道名 隊列名} 81 return self.info 82 83 def retf(self,str_l):#反射命令 84 sl=str_l.split()[0]#取命令開頭 85 if sl==‘helps‘: 86 self.helps() 87 if len(str_l.split())==1 and sl!=‘check_task_all‘ : 88 return 89 if hasattr(self,sl):#是否存在 90 func=getattr(self,sl)#調用 91 rer=func(str_l)#執行 92 #print(rer) 93 if rer is not None: 94 for i in rer: 95 print("任務id:%s"%i) 96 97 def attr_l(self,n):#命令分解函數 98 attr=n.split("--")##用--分割 99 addr=attr[1].split()[1:]#獲取IP列表 100 return addr#返回IP列表 101 102 def oreds_(self,n):#獲取 命令 103 oreds=n.split("\"")[1]##用"分割取命令 104 return oreds#返回 命令 105 106 def helps(self):#查看指令幫助 107 print(self.help_info) 108 109 def th_start(self):#開始 110 self.helps() 111 while True: 112 str_l=input(">>:").strip() 113 if not str_l:continue#如果為空重新輸入 114 t1=threading.Thread(target=self.retf,args=(str_l,))#創建新線程 調用反射函數 115 t1.start()#開始線程View Code
|- - -PRC_SERVER/#服務端程序目錄
| |- - -__init__.py
| |- - -bin/#執行目錄
| | |- - -__init__.py
| | |- - -server_start.py#服務端程序執行文件
1 import os ,sys 2 BASE_DIR=os.path.dirname(os.path.dirname(os.path.abspath(__file__)))#獲取相對路徑轉為絕對路徑賦於變量 3 sys.path.append(BASE_DIR)#增加環境變量 4 5 from core.client_class import Threa 6 7 if __name__ == ‘__main__‘: 8 RPCS=Threa() 9 response=RPCS.th_start()View Code
| |- - -core/##主邏輯程序目錄
| | |- - -server_class.py#主邏輯 相關類
1 import pika,os 2 3 class RabbitMQ_PRC(object): 4 def __init__(self,myaddr): 5 self.queues=myaddr#用本機IP做隊列名 6 self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘localhost‘))#生成消息對隊 7 self.channel = self.connection.channel()#生成管道 8 self.channel.queue_declare(queue=self.queues)#消息收接隊列 9 10 def str_run(self,body):#處理 run的函數 11 msg = os.popen(body.decode()).read()#執行系統命令 12 if not msg: 13 msg = ‘系統命令不存在‘ 14 return msg 15 16 def on_request(self,ch, method, props, body):#回調函數 17 resp=self.str_run(body) 18 print(‘執行完成‘) 19 #print(resp) 20 ch.basic_publish(exchange=‘‘, 21 routing_key=props.reply_to,#收消息的隊列 22 properties=pika.BasicProperties(correlation_id =props.correlation_id),#返回消息的隊列 23 body=str(resp))#返回結果數據 24 ch.basic_ack(delivery_tag = method.delivery_tag)##確保消息被 客戶端接收 25 26 def run_(self): 27 self.channel.basic_qos(prefetch_count=1)#同時只處理一個消息 28 self.channel.basic_consume(self.on_request, queue=self.queues)#接收消息，自動調用回調函數 29 30 print("開始接收數據！") 31 self.channel.start_consuming()#開始接收View Code
python第六十三天-- 第十一周作業