基于RabbitMQ的direct任务驱动异步RPC程序实现：
RPC_dispatcher指令分发器：
1 #!/usr/bin/env python
2 # -*- coding:utf-8 -*-
3 __Author__ = "Zhang Xuyao"
4
5 import pika
6 import uuid
7 import time
8 import threading
9
10
11 class RpcDispatcher(object):
12 def __init__(self, rMQ_addr):
13 self.cmd_list = [
14 "ls", "df", "free",
15 "ip", "ifconfig", "tail",
16 "head", "grep", "uptime",
17 "date"
18 ]
19 self.task_dict = {}
20 self.task_fin_dict = {}
21 self.routing_key_list = []
22
23 self.connection = pika.BlockingConnection(
24 pika.ConnectionParameters(host=rMQ_addr)
25 )
26 self.channel = self.connection.channel()
27
28 # 收数据用
29 recv_queue = self.channel.queue_declare(exclusive=True)
30 self.recv_queue_name = recv_queue.method.queue
31 self.channel.basic_consume(self._on_response, queue=self.recv_queue_name)
32
33 # 发数据用
34 self.channel.exchange_declare(exchange=‘send‘, type=‘direct‘)
35 self.send_queue = self.channel.queue_declare()
36 self.send_queue_name = self.send_queue.method.queue
37
38 # 获取指定通道的响应(4)###
39 def _on_response(self, ch, method, parameters, msg):
40 if parameters.correlation_id in self.task_dict:
41 self.task_dict[parameters.correlation_id][parameters.app_id]["response"] = msg.decode("utf-8")
42 ch.basic_ack(delivery_tag=method.delivery_tag)
43 self.task_dict[parameters.correlation_id][parameters.app_id]["recv_time"] = time.time()
44 fin_flag = False
45 for host in self.task_dict[parameters.correlation_id]:
46 if self.task_dict[parameters.correlation_id][host]["response"] != None:
47 continue
48 else:
49
50 break
51 # 执行结果全部收到就把记录转移到已经完成容器中，根据检查已经完成容器中的内容用于确定任务的状态
52 else:
53 self.task_fin_dict[parameters.correlation_id] = self.task_dict[parameters.correlation_id]
54 del self.task_dict[parameters.correlation_id]
55
56 # 发送请求（2）######
57 def _on_request(self, input_cmd, host_list):
58 print(RpcDispatcher.colorStr("[x]Requesting>>: ‘%s‘ on %s"
59 % (cmd, tuple(host_list)), 33))
60 self.response = None
61 # 生成全局校验码
62 corr_id = str(uuid.uuid4())
63 print(RpcDispatcher.colorStr("[x]Task_id>>: %s"
64 % corr_id, 34))
65 self.task_dict[corr_id] = {}
66 if host_list:
67 for host in host_list:
68 self.task_dict[corr_id][host] = {
69 "cmd": input_cmd,
70 "response": None,
71 "req_time": time.time(),
72 "recv_time": None,
73 }
74 # 绑定routing_key准备并发布消息
75 self.channel.queue_bind(exchange=‘send‘,
76 queue=self.send_queue_name,
77 routing_key=host)
78 # 向执行器并发布消息指令
79 self.channel.basic_publish(exchange=‘send‘,
80 routing_key=host,
81 properties=pika.BasicProperties(
82 reply_to=self.recv_queue_name,
83 correlation_id=corr_id,
84 ),
85 body=str(input_cmd))
86 # 消息发布后解除routing_key绑定
87 self.channel.queue_unbind(exchange=‘send‘,
88 queue=self.send_queue_name,
89 routing_key=host)
90
91 # 守护线程负责不断检测响应结果是否全部收到
92 on_recv_thread = threading.Thread(target=self._on_recv, args=[corr_id, ])
93 on_recv_thread.setDaemon(True)
94 on_recv_thread.start()
95
96 # 等待数据消息(3)
97 def _on_recv(self, task_id):
98 # 根据检查已经完成容器中的指定task_id是否存在来确定任务的状态，为空则继续收取消息
99 while task_id not in self.task_fin_dict:
100 self.connection.process_data_events()
101
102 # 显示已经完成的任务编号（5）
103 def show_task_fin(self):
104 print("尚未查看的任务：")
105 for task_id in self.task_fin_dict:
106 value_list = tuple(self.task_fin_dict[task_id].values())
107 host_list = tuple(self.task_fin_dict[task_id].keys())
108 cmd = str(value_list[0]["cmd"])
109 print(RpcDispatcher.colorStr("[task_id]: %s | [cmd_info]: ‘%s‘ | [host_list]: %s"
110 % (task_id, cmd, host_list), 32))
111
112 # 获取指定任务的执行结果（6）
113 def get_response(self, task_id):
114 if task_id in self.task_fin_dict:
115 for host in self.task_fin_dict[task_id]:
116 response = self.task_fin_dict[task_id][host]["response"]
117 cmd_req = self.task_fin_dict[task_id][host]["cmd"]
118 time_cost = self.task_fin_dict[task_id][host]["recv_time"] - 119 self.task_fin_dict[task_id][host]["req_time"]
120 time_cost = round(time_cost, 3)
121 print(RpcDispatcher.colorStr("Host: %s | Cmd: ‘%s‘ \nTime Cost: %ss | Response: "
122 % (host, cmd_req, time_cost), 33))
123 print(RpcDispatcher.colorStr(response, 36))
124 del self.task_fin_dict[task_id]
125 else:
126 print("任务结果尚未全部返回")
127
128 # 接收外部输入，调用请求（1）
129 def call(self, cmd, host_list):
130 return self._on_request(cmd, host_list)
131
132 @staticmethod
133 def colorStr(aStr, color_code):
134 return "\033[0;" + str(color_code) + ";0m" + aStr + "\033[0m"
135
136 def __del__(self):
137 self.connection.close()
138
139
140 if __name__ == ‘__main__‘:
141 cmd_rpc = RpcDispatcher(rMQ_addr="localhost")
142 while True:
143 cmd = input("[$]Cmd>>:")
144 if cmd.lower() != ‘eof‘ and cmd.lower() != ‘exit‘:
145 if cmd.split()[0].lower() in ["$s"]:
146 cmd_rpc.show_task_fin()
147 elif cmd.split()[0].lower() in ["$c"] and len(cmd.split()) == 2:
148 cmd_rpc.get_response(cmd.split()[1])
149 else:
150 cmd_split = cmd.split()
151 has_host = cmd_split.count("--hosts")
152 if has_host != 1:
153 print(cmd_rpc.colorStr("Usage <cmd> --hosts <ip1>[,<ip2>[,<ip3>...]]", 35))
154 continue
155 else:
156 if len(cmd_split) <= cmd_split.index("--hosts") + 1:
157 print("请至少指定一个主机IP")
158 continue
159 host_list = cmd_split[cmd_split.index("--hosts") + 1].split(‘,‘)
160 cmd = " ".join(cmd_split[0:cmd_split.index("--hosts")]).strip()
161
162 if cmd.split()[0] in cmd_rpc.cmd_list:
163 cmd_rpc.call(cmd, host_list)
164 else:
165 print("您输入的命令暂不支持...")
166 continue
167 else:
168 break
RPC 指令分发器代码
RPC_executor指令执行器：
1 #!/usr/bin/env python
2 # -*- coding:utf-8 -*-
3 __Author__ = "Zhang Xuyao"
4
5 import pika
6 import os, time
7
8
9 class RpcExecutor(object):
10 def __init__(self, rMQ_addr, ip):
11 self.connection = pika.BlockingConnection(pika.ConnectionParameters(
12 host=rMQ_addr))
13 self.channel = self.connection.channel()
14
15 # 接收命令用
16 self.ip = ip
17 self.channel.exchange_declare(exchange=‘send‘, type=‘direct‘)
18 self.send_queue = self.channel.queue_declare()
19 self.send_queue_name = self.send_queue.method.queue
20 self.channel.queue_bind(exchange=‘send‘, routing_key=self.ip, queue=self.send_queue_name)
21 self.channel.basic_consume(self._on_request, queue=self.send_queue_name)
22
23 # 开始订阅消息
24 def run(self):
25 print(" [x] Awaiting RPC requests")
26 self.channel.start_consuming()
27
28 # 执行消息命令
29 def exec_cmd(self, cmd_str):
30 result = os.popen(cmd_str).read()
31 if not result:
32 return "命令没有输出结果"
33 else:
34 return result
35
36 # 请求事件的回调函数
37 def _on_request(self, ch, method, props, body):
38 cmd_str = body.decode(‘utf-8‘)
39 print(" [.] exec_cmd(%s)" % cmd_str)
40 response = self.exec_cmd(cmd_str)
41
42 # 发送命令结果，通过传来的queue进行发布，这里的app_id是额外增加的IP信息，区分于其他执行器的响应结果
43 ch.basic_publish(exchange=‘‘,
44 routing_key=props.reply_to,
45 properties=pika.BasicProperties(
46 correlation_id=props.correlation_id,
47 app_id=self.ip
48 ),
49 body=str(response)
50 )
51 # 消息反馈确认，确保对方确实收到了响应结果
52 ch.basic_ack(delivery_tag=method.delivery_tag)
53
54 def __del__(self):
55 self.connection.close()
56
57
58 if __name__ == ‘__main__‘:
59 ip = input(‘请输入IP地址>>:‘)
60 cmd_executor = RpcExecutor(‘localhost‘, ip)
61 cmd_executor.run()
RPC 执行器代码
时间： 2024-10-29 04:26:18