基於RabbitMQ RPC實現的主機異步管理
阿新 • • 發佈：2018-01-26
fff 批量操作 result sel lis llb use exchange port README:
1、需求 - [ ] 利用RibbitMQ進行數據交互 - [ ] 可以對多臺服務器進行批量操作 - [ ] 執行命令後不等待命令的執行結果，而是直接讓輸入下一條命令，結果出來後自動打印 - [ ] 實現異步操作 備註 - [ ] RabbitMQ隊列： ①執行命令時，隊列名為“rpc_queue2” ②查詢數據時，用的是回調時隨機生成的callback_queue名 ③conf/settings——Rabbitmq地址“192.168.17.102”，端口：5672，用戶名：admin，密碼：admin - [ ] SSH: RPC_Server/server.py——paramiko操作連接的測試Linux默認端口22,用戶名：root，密碼：123456 - [ ] threading多線程： 實現命令執行後不等待執行結果，依然可以輸入新的指令 - [ ] 執行命令格式： -->>run ifconfig host 192.168.20.22 192.168.20.23 dir server端要執行的命令 host host後可跟一個或多個可以通過rabbitMQ的服務器地址 - [ ] 查看後臺所有的TASK_ID信息： -->>check_task 顯示結果樣式：TASK_ID【76786】 HOST【192.168.20.22】 COMMAND【dir】 TASK_ID【10307】 HOST【192.168.20.23】 COMMAND【dir】 - [ ] 查看TASK_ID對應的執行結果： -->>get_task 10307 程序目錄結構： ├── README.md ├── RPC_Client │?? ├── bin │?? │?? ├── __init__.py │?? │?? └── start.py #客戶端啟動程序 │?? ├── conf │?? │?? ├── __init__.py │?? │?? ├── __pycache__ │?? │?? │?? ├── __init__.cpython-36.pyc │?? │?? │?? └── settings.cpython-36.pyc │?? │?? └── settings.py │?? ├── core │?? │?? ├── __init__.py │?? │?? ├── main.py │?? │?? └── __pycache__ │?? │?? ├── __init__.cpython-36.pyc │?? │?? └── main.cpython-36.pyc │?? └── modules │?? ├── client.py │?? ├── __init__.py │?? └── __pycache__ │?? ├── client.cpython-36.pyc │?? └── __init__.cpython-36.pyc └── RPC_Server ├── conf │?? ├── __pycache__ │?? │?? └── settings.cpython-36.pyc │?? └── settings.py └── server.py #server端啟動程序 程序啟動： 客戶端啟動：RPC_Client/bin/start.py 服務端啟動：RPC_Server/server.py
RPC 客戶端bin目錄start.py
import os,sys,platform if platform.system() == ‘Windows‘: BASE_DIR = ‘\\‘.join(os.path.abspath(os.path.dirname(__file__)).split(‘\\‘)[:-1]) else: BASE_DIR = ‘/‘.join(os.path.abspath(os.path.dirname(__file__)).split(‘/‘)[:-1]) sys.path.append(BASE_DIR) from core import main if __name__ == "__main__": handle = main.Handle() handle.start()
RPC客戶度conf目錄 settings.py
import os,sys,platform if platform.system() == ‘Windows‘: BASE_DIR = ‘\\‘.join(os.path.abspath(os.path.dirname(__file__)).split(‘\\‘)[:-1]) else: BASE_DIR = ‘/‘.join(os.path.abspath(os.path.dirname(__file__)).split(‘/‘)[:-1]) sys.path.append(BASE_DIR) RabbitmqHost = ‘192.168.17.102‘ RabbitmqUser = ‘admin‘ RabbitmqPwd = ‘admin‘ credentails = pika.PlainCredentials(RabbitmqUser,RabbitmqPwd)
RPC客戶端主程序core/main.py
import pika
import random
import threading
from modules import client
from conf import settings
class Handle(object):
def __init__(self):
# 建立連接，指定服務器的ip地址
self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=settings.RabbitmqHost,credentials=settings.credentails))
# 建立一個會話，每個channel代表一個會話任務
self.channel = self.connection.channel()
def run_cmd(self,cmd, host):
rpc_client = client.Client(self.connection,self.channel)
task_id = str(random.randint(1000,9999)) #生成4位的Correlation id
response = rpc_client.call(cmd, host)
self.corr_id = response[1]
print(‘Task_ID‘,task_id)
self.info[task_id] = [self.corr_id,host,cmd,response[0],response[1]]
def start(self):
self.info = {} #task 返回任務信息字典
help = ‘‘‘
命令格式
執行系統命令：run command host eg: run ls 10.10.0.10
查看所有執行任務：check_task
查看指定任務結果：get_task id eg:get_task 6723
‘‘‘
print(help)
while True:
msg = input(">> ").strip()
if msg.startswith(‘run‘) and len(msg.split()) >= 3:
cmd = msg.split()[1]
#多線程運行
th_join = []
for host in msg.split()[2:]:
th = threading.Thread(target=self.run_cmd,args=(cmd,host,))
th.start()
th_join.append(th)
for t in th_join:
t.join()
elif msg == ‘check_task‘:
if not self.info:
print("沒有任務隊列")
continue
else:
for taskid,task in self.info.items():
print(‘TaskID [%s] Host [%s] COMMAND [%s]‘%(taskid,task[1],task[2]))
elif msg.startswith(‘get_task‘):
rpc_client = client.Client(self.connection,self.channel)
if msg.split()[1] in self.info:
task_id = msg.split()[1]
callback_queue = self.info[task_id][3]
correlation_id = self.info[task_id][4]
task_result = rpc_client.get_task(callback_queue,correlation_id)
del self.info[task_id]
print(task_result.decode().strip())
else:
print(‘輸入的task ID不存在！‘)
continue
elif not msg:
continue
else:
print(‘輸入錯誤，請重新輸入！‘)
continueRPC客戶端modules
import pika
import random
import uuid
class Client(object):
def __init__(self,connection,channel):
self.connection = connection
self.channel = channel
# 對回調隊列中的響應進行處理的函數
def on_response(self, ch, method, props, body):
if self.correlation_id == props.correlation_id:
self.response = body
ch.basic_ack(delivery_tag=method.delivery_tag)
def get_task(self,callback_queue,correlation_id):
# 初始化 response
self.response = None
self.correlation_id = correlation_id
# 客戶端訂閱回調隊列，當回調隊列中有響應時，調用`on_response`方法對響應進行處理;
self.channel.basic_consume(self.on_response,queue=callback_queue)
while self.response is None:
self.connection.process_data_events()
return self.response
# 發出RPC請求
def call(self,cmd,host):
# 聲明回調隊列，再次聲明的原因是，服務器和客戶端可能先後開啟，該聲明是冪等的，多次聲明，但只生效一次
result = self.channel.queue_declare(exclusive=True)
# 將次隊列指定為當前客戶端的回調隊列
self.callback_queue = result.method.queue
msg = cmd + " " + "".join(host)
self.corr_id = str(uuid.uuid4())
#self.corr_id = corr_id
# 發送RPC請求內容到RPC請求隊列`rpc_queue`，同時發送的還有`reply_to`和`correlation_id`
self.channel.basic_publish(exchange=‘‘,
routing_key=‘rpc_queue2‘,
properties=pika.BasicProperties(
reply_to=self.callback_queue,
correlation_id=self.corr_id,
),
body=msg)
return self.callback_queue,self.corr_id
RPC服務器settings.py
RabbitmqHost = ‘192.168.17.102‘
RabbitmqUser = ‘admin‘
RabbitmqPwd = ‘admin‘
credentails = pika.PlainCredentials(RabbitmqUser,RabbitmqPwd)RPC服務端主程序：
#!/usr/bin/env python
import pika
import paramiko
import os,sys,platform
if platform.system() == ‘Windows‘:
BASE_DIR = ‘\\‘.join(os.path.abspath(os.path.dirname(__file__)).split(‘\\‘)[:-1])
else:
BASE_DIR = ‘/‘.join(os.path.abspath(os.path.dirname(__file__)).split(‘/‘)[:-1])
sys.path.append(BASE_DIR)
from conf import settings
# 建立連接，服務器地址為localhost，可指定ip地址
connection = pika.BlockingConnection(pika.ConnectionParameters(host=settings.RabbitmqHost,credentials=settings.credentails))
# 建立會話
channel = connection.channel()
# 聲明RPC請求隊列
channel.queue_declare(queue=‘rpc_queue2‘)
# 數據處理方法
def exec_cmd(cmd,host):
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(hostname=host, port=22, username=‘root‘, password=‘123456‘,timeout=10)
stdin, stdout, stderr = ssh.exec_command(cmd)
stdout_result = stdout.read()
stderr_result = stderr.read()
result = stdout_result if stdout_result else stderr_result
return result.decode()
ssh.close()
# 對RPC請求隊列中的請求進行處理
def on_request(ch, method, props, body):
cmd = body.split()[0]
host = body.split()[1]
# 調用數據處理方法
response = exec_cmd(cmd,host)
# 將處理結果(響應)發送到回調隊列
ch.basic_publish(exchange=‘‘,
routing_key=props.reply_to,
properties=pika.BasicProperties(correlation_id = props.correlation_id),
body=str(response))
ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_consume(on_request, queue=‘rpc_queue2‘)
print(" [x] Awaiting RPC requests")
channel.start_consuming()基於RabbitMQ RPC實現的主機異步管理