RPC，远程连接Linux开发

标签：__ self queue RPC Linux print data 远程 def

RPC

指远程过程调用

eg：

两台服务器，A ，B， A想要调用B服务器上的应用的函数或方法，但是他两不在一个内存空间，不能直接调用，需要通过网络来表达调用的语义和传达调用的数据，只要调用成功这就称为RPC调用.

主要应用在分布式系统当中与微服务。

由于计算能力需要横向扩展，需要在多台机器组成的集群上部署应用，可以在不同语言中调用。

集群与分布式的区别

集群：在多台机器上部署一个项目

分布式：在多台机器上部署多个服务组成一个项目

常见的RPC框架

dubbo Java上用的只支持java与java通信

gRpc 谷歌开源的，支持跨语言。用go写的通过python来调用

python实现RPC

SimpleXMLRPCServer python自带的模块

ZeroRPC 第三方模块

SimpleXMLRPCServer实现

server.py

from xmlrpc.server import SimpleXMLRPCServer

# 通信格式使用xml

class RPCserver(object):
    def __init__(self):
        super(RPCserver,self).__init__()
        print(self)
        self.send_data = 'lxj nb'
        self.recv_data = None

    def getObj(self):
        print('get_obj')
        return self.send_data

    def sendObj(self,data):
        print('send_obj')
        self.recv_data = data
        print(self.recv_data)

server = SimpleXMLRPCServer(('localhost', 8080),allow_none=True)
server.register_introspection_functions()
server.register_instance(RPCserver())
# 启动服务监听函数
server.serve_forever()

client.py

import time
from xmlrpc.client import ServerProxy

def xmlrpc_client():
    print('xmlrpc_client')
    c = ServerProxy('http://localhost:8080')

    res = c.getObj()
    print(res)
    res1 = c.sendObj('lxj sgdsb')
    print(res1)

    # 案例
    # data = 'lxj nb'
    # start=time.time()
    # for i in range(500):
    #     a=c.getObj()
    #     print(a)
    # for i in range(500):
    #     c.sendObj(data)
    # print(time.time()-start)

if __name__ == '__main__':
    xmlrpc_client()

ZeroRPC来实现

比SimpleXMLRPCServer 速度快，本质是用了一个消息队列

server.py

import zerorpc

class RPCServer(object):

    def __init__(self):
        super(RPCServer, self).__init__()
        print(self)
        self.send_data = 'lqz nb'
        self.recv_data = None

    def getObj(self):
        print('get data')
        return self.send_data

    def sendObj(self, data):
        print('send data')
        self.recv_data = data
        print(self.recv_data)
# zerorpc
s = zerorpc.Server(RPCServer())
s.bind('tcp://0.0.0.0:4243')
s.run()

client.py

import zerorpc
import time
# zerorpc
def zerorpc_client():
    print('zerorpc client')
    c = zerorpc.Client()
    c.connect('tcp://127.0.0.1:4243')
    data = 'lqz nb'

    start = time.time()
    for i in range(500):
        a=c.getObj()
        print(a)
    for i in range(500):
        c.sendObj(data)

    print('total time %s' % (time.time() - start))


if __name__ == '__main__':
    zerorpc_client()

rabbitMQ实现RPC

Client.py

import pika
import uuid


class FibonacciRpcClient(object):
    def __init__(self):
        credentials = pika.PlainCredentials("admin", "admin")
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('10.0.0.101', credentials=credentials))
        self.channel = self.connection.channel()

        # 随机生成一个消息队列(用于接收结果)
        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue

        # 监听消息队列中是否有值返回,如果有值则执行 on_response 函数(一旦有结果,则执行on_response)
        self.channel.basic_consume(queue=self.callback_queue, on_message_callback=self.on_response, auto_ack=True)

    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())

        # 客户端 给 服务端 发送一个任务:  任务id = corr_id / 任务内容 = '30' / 用于接收结果的队列名称
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue',  # 服务端接收任务的队列名称
                                   properties=pika.BasicProperties(
                                       reply_to=self.callback_queue,  # 用于接收结果的队列
                                       correlation_id=self.corr_id,  # 任务ID
                                   ),
                                   body=str(n))
        while self.response is None:
            self.connection.process_data_events()

        return self.response


fibonacci_rpc = FibonacciRpcClient()

response = fibonacci_rpc.call(9)
print('返回结果:', response)

server.py

import pika
credentials = pika.PlainCredentials("admin","admin")
connection = pika.BlockingConnection(pika.ConnectionParameters('10.0.0.101',credentials=credentials))
channel = connection.channel()

# 声明一个队列rpc_queue
channel.queue_declare(queue='rpc_queue')

def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n - 1) + fib(n - 2)

def on_request(ch, method, props, body):
    n = int(body)

    print(" [.] fib(%s)" % n)
    response = fib(n)

    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_qos(prefetch_count=1)
channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)

print(" [x] Awaiting RPC requests")
channel.start_consuming()