那些年被我坑过的Python——第十章Broker（rabbitMQ/redis）

Posted 2020-08-15

基于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 执行器代码