python项目开发：用RabbitMQ实现异步RPC

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了python项目开发：用RabbitMQ实现异步RPC，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含5797字，纯文字阅读大概需要9分钟。

内容图文

程序要求：

              1. 用Rabbit MQ实现RPC

              1. 可以异步地执行多条命令

              2. 可以对一次性对多个机器执行命令

程序效果：

              ---》run dir host1 host2 。。。。

              ---》get task_id

              ---》taskId：xxxx   host： xxxxxx

              ---》check task_id

              --->打印结果

程序分析：

              为了达到异步地效果，可以使用多线程或协程，即每执行一条命令就启动一条线程或协程。客户端发送命令到队列、从返回队列接收结果分离，不能写到一起。

业务逻辑：

代码实现：

README

#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_id
 3. check task_id #获取命令结果
 4. delete task_id #删除命令结果

View Code

RPC_server\\bin\\start.py

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()

View Code

RPC_server\\core\\RpcServer.py

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.queue
        self.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

View Code

RPC_client\\bin\\start.py

import sys,os

BASE_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()

View Code

RPC_client\\core\\main.py

import random
import threading
from modules import RpcClient

class run(object):
    def __init__(self):
        self.client = RpcClient.RpcClient()
        self.information = {}

    def start(self):
        while True:
            try:
                command = input("-->")
                if not command:continue
                t = 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))
        except 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)
    def delete(self,command):
        '''删除命令结果'''
        try:
            for task_id in command.split()[1:]:
                self.information.pop(task_id)
        except Exception as e:
            print(e)

View Code

RPC_client\\conf\\settings.py

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

View Code

RPC_client\\mudules\\RpcClient.py

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_id
        self.response = None
        self.channel.basic_consume(
            self.on_response,
            queue=callback_queue,
        )
        while self.response is None:
            self.connection.process_data_events() #非阻塞版的start_consuming
        return self.response

    def 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 #返回queue
        corr_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

View Code

程序执行实例：

内容总结

以上是互联网集市为您收集整理的python项目开发：用RabbitMQ实现异步RPC全部内容，希望文章能够帮你解决python项目开发：用RabbitMQ实现异步RPC所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

---