Python 程序：基于RabbitMQ实现主机管理

Python 程序：基于RabbitMQ实现主机管理

---

1、需求

2、代码

3、测试样图

---

一、需求

1、可以异步的执行多个命令

2、对多台机器

举例

>>:run "df -h" --hosts 192.168.3.55 10.4.3.4 
task id: 45334
>>: check_task 45334 
>>:

二、代码

import random
import pika

class rpc_cilent(object):
    def connect(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    def connect_1(self):
        self.credentials = pika.PlainCredentials('zz', '123456')
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(
        '192.168.43.165',5672,'/',self.credentials))
    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body
    def call(self, command, host,tmp_dict):
        if host == "127.0.0.1":
            self.connect()
        elif host == "192.168.43.165":
            self.connect_1()
        self.channel = self.connection.channel()
        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue
        self.channel.basic_consume(self.on_response, no_ack=True,queue=self.callback_queue)
        self.response = None
        self.corr_id = str(random.randint(10000, 99999))
        ack = self.channel.basic_publish(exchange='',
                                         routing_key= "127.0.0.1",
                                         properties=pika.BasicProperties(
                                             reply_to=self.callback_queue,
                                             correlation_id=self.corr_id),
                                         body=str(command))
        while self.response is None:
            self.connection.process_data_events()
        task_id = self.corr_id
        res = self.response.decode()
        tmp_dict[task_id] = res
        print('33[42;0mtask_id: %s host: %s cmd: %s33[0m' % (self.corr_id, host, command))
class rpc_start(object):
    def __init__(self):
        self.tmp_dict = {}
        self.start()
    def check_all(self,*args):
        for index, key in enumerate(self.tmp_dict.keys()):
            print(index, '33[42;0mtask_id: %s33[0m' % key)
    def check_task(self,user_cmd):
        try:
            command_list = user_cmd.split()
            print(self.tmp_dict[command_list[1]])
            del self.tmp_dict[command_list[1]]
        except IndexError:
            help()
    def run(self,user_cmd):
         try:
            hosts_obj = ( user_cmd.split('hosts'))
            hosts = hosts_obj[1].strip().split()
            command = user_cmd.split(""")[1]
            for host in hosts:
                try:
                    if host == "127.0.0.1":
                        rpc_cilent.call(command, host,self.tmp_dict)
                    elif host == "192.168.43.165":
                        rpc_cilent.call(command, host,self.tmp_dict)
                    else:
                        print("33[41;0mno host:%s33[0m"%host)
                except TypeError and AssertionError:
                    break
         except IndexError:
            print("33[31;0mcommand not found33[0m")
            self.help()
    def help(self):
        print('33[34;0mUsage: run "df -h" hosts 127.0.0.1 192.168.43.165 33[0m')
        print('33[34;0m       check_task 9712833[0m')
        print('33[34;0m       check_all33[0m')
    def start(self):
        self.help()
        while True:
            user_cmd = input("->>").strip()
            if not user_cmd:continue
            self.cmd = user_cmd.split()[0]
            if hasattr(self, self.cmd):
                getattr(self, self.cmd)(user_cmd)
            else:
                print("33[31;0mcommand not found!33[0m")
                self.help()

rpc_cilent = rpc_cilent()
start = rpc_start()

import pika
import subprocess

class rpc_server(object):
    def __init__(self,host,queue):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=host))
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue=queue)
        self.channel.basic_qos(prefetch_count=1)
        self.channel.basic_consume(self.on_request, queue=queue)
        print(" [x] Awaiting RPC requests")
        self.channel.start_consuming()
    def command(self,cmd):
        res = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        msg = res.stdout.read().decode('gbk')
        if len(msg) == 0:
            msg = res.stderr.read().decode('gbk')
        print(msg)
        return msg
    def on_request(self,ch, method, props, body):
        cmd = body.decode()
        print(cmd)
        respone = self.command(cmd)
        ch.basic_publish(exchange='',
                         routing_key=props.reply_to,
                         properties=pika.BasicProperties(correlation_id=props.correlation_id),
                         body=respone)
        ch.basic_ack(delivery_tag=method.delivery_tag)

server = rpc_server("localhost","127.0.0.1")

三、测试样图

启动客户端

windows端服务器启动

linux端服务器启动

client端测试：