#author：Wu zhiHao#博客地址：https://www.cnblogs.com/BUPT-MrWu/p/10364619.html#程序目录框架:|--RPC|--RPC_server #服务端|--bin|--start.py #程序入口|--core|--RpcServer.py #服务端主要逻辑|--RPC_client #客户端|--bin|--start.py #程序入口|--core|--main.py #程序主要逻辑|--modules|--RpcClient.py #客户端主要逻辑|--conf|--settings.py #配置文件|--READ_ME#命令格式:1. run command host1 host2..... #执行命令2. all_task #获取全部task_id3. check task_id #获取命令结果

import sys,os
BASE_dir = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
sys.path.append(BASE_dir)
from core import RpcServer
if __name__ == '__main__':obj = RpcServer.RpcServer()obj.channel.start_consuming()

import pika
import os
import socket
from conf import settings
class RpcServer(object):def __init__(self):self.credentials = pika.PlainCredentials(settings.RabbitMq_name,settings.RabbitMq_password) #RabbiMQ用户认证self.connection = pika.BlockingConnection(pika.ConnectionParameters(settings.RabbitMq_ip,settings.RabbitMq_port,"/",self.credentials,))self.My_Ip = self.get_ip() #获取服务端IP地址self.channel = self.connection.channel()self.result = self.channel.queue_declare(exclusive=True)self.queue_name = self.result.method.queueself.channel.exchange_declare(exchange="Rpc",exchange_type="direct",)self.channel.queue_bind(exchange="Rpc",queue=self.queue_name,routing_key=self.My_Ip,)self.channel.basic_consume(self.on_response,queue=self.queue_name,)def on_response(self,ch,method,properties,body):command = body.decode()command_result = self.on_request(command)self.channel.basic_publish(exchange="",routing_key=properties.reply_to,properties=pika.BasicProperties(correlation_id=properties.correlation_id,),body=command_result)def on_request(self,command):return os.popen(command).read()def get_ip(self):computer_name = socket.getfqdn(socket.gethostname(  ))computer_Ip = socket.gethostbyname(computer_name)return computer_Ip

import sys,osBASE_dir = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
sys.path.append(BASE_dir)
from core import main
if __name__ == '__main__':obj = main.run()obj.start()

import random
import threading
from modules import RpcClientclass run(object):def __init__(self):self.client = RpcClient.RpcClient()self.information = {}def start(self):while True:try:command = input("-->")if not command:continuet = threading.Thread(target=self.select,args=(command,))t.start()except Exception as e:print(e)def select(self,command):'''解析命令'''try:keyword = command.split()[0]func = getattr(self,keyword)func(command)except Exception as e:print(e)def run(self,command):'''执行命令'''try:task_id = str(random.randint(100,1000))self.information[task_id] = {}keyword = command.split()[1]for host in command.split()[2:]:result = self.client.on_request(host,keyword)self.information[task_id][host] = [result[0],result[1]]except Exception as e:print(e)def check(self,command):'''获取命令结果'''try:task_id = command.split()[1]for host in self.information[task_id]:corr_id = self.information[task_id][host][0]callback_queue = self.information[task_id][host][1]command_result = self.client.get_response(corr_id,callback_queue)print("%s:\n%s"%(host,command_result))self.information.pop(task_id) #删除task_idexcept Exception as e:print(e)def all_task(self,command):'''获取全部task_id'''try:for task_id in self.information:all_host = []for host in self.information[task_id]:all_host.append(host)print("task_id: %s  host: %s\n"%(task_id,all_host))except Exception as e:print(e)

RabbitMq_name = "XXX" #RabbitMq用户名
RabbitMq_password = "XXX" #rabbitmq用户密码
RabbitMq_ip = "XXX" #RabbitMq端的IP地址
RabbitMq_port = 5672 #RabbitMq端的端口号

import pika
import uuid
from conf import settings
class RpcClient(object):def __init__(self):self.credentials = pika.PlainCredentials(settings.RabbitMq_name,settings.RabbitMq_password) #RabbiMQ用户认证self.connection = pika.BlockingConnection(pika.ConnectionParameters(settings.RabbitMq_ip,settings.RabbitMq_port,"/",self.credentials,))self.channel = self.connection.channel()def get_response(self,corr_id,callback_queue):'''从队列里取值'''self.corr_id = corr_idself.response = Noneself.channel.basic_consume(self.on_response,queue=callback_queue,)while self.response is None:self.connection.process_data_events() #非阻塞版的start_consumingreturn self.responsedef on_response(self,ch,method,properties,body):'''当队列里有数据时执行'''if self.corr_id == properties.correlation_id:self.response = body.decode()def on_request(self,host,command):'''发送命令'''result = self.channel.queue_declare(exclusive=False) #生成另一个queue时，这个queue不会消失callback_queue = result.method.queue #返回queuecorr_id = str(uuid.uuid4()) #验证码
        self.channel.exchange_declare(exchange="Rpc",exchange_type="direct")self.channel.basic_publish(exchange="Rpc",routing_key=host,properties=pika.BasicProperties(correlation_id=corr_id,reply_to=callback_queue,),body=command,)return corr_id,callback_queue #返回验证值和返回queue