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()#开始线程