远程过程调用

2021-09-12 00:42:35阅读 7评论 0

简单实现

服务端

from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCServer

def add(a,b):
    return a + b

server = SimpleJSONRPCServer(('127.0.0.1',8000))
server.register_function(add)
print("Listen on port 8000")
server.serve_forever()

客户端

import jsonrpclib

server = jsonrpclib.ServerProxy('http://127.0.0.1:8000')
print(server.add(1,2))

考虑问题

1）超时机制
2）限流，处于长期高可用状态
3）解耦
4）负载均衡，微服务-分布式应用的具体体现
5）序列化和饭序列化是否高效

zerorpc

简单调用

服务端

import zerorpc

class Caculator:
    def add(self,a,b):
        return a + b

s = zerorpc.Server(Caculator())
s.bind('tcp://0.0.0.0:8000')
s.run()

客户端

import zerorpc

c = zerorpc.Client()
c.connect('tcp://127.0.0.1:8000')
print(c.add(1,2))

流式调用

服务端

import zerorpc

class StreamingRPC(object):
    @zerorpc.stream
    def streaming_range(self, fr, to, step):
        return range(fr, to, step)

s = zerorpc.Server(StreamingRPC())
s.bind("tcp://0.0.0.0:4242")
s.run()

客户端

import zerorpc

c = zerorpc.Client()
c.connect("tcp://127.0.0.1:4242")

for item in c.streaming_range(10, 20, 2):
    print(item)

go语言实现

简单实现

服务端

package main

import (
    "net"
    "net/rpc"
)

type HelloService struct {}

func (s *HelloService) Hello(arg string,reply *string)error{
    *reply =  "hello" + arg
    return nil
}

func main()  {
    listener,_ := net.Listen("tcp",":8000")
    _ = rpc.RegisterName("HelloService",&HelloService{})
    conn,_ := listener.Accept()
    rpc.ServeConn(conn)
}

客户端

package main

import "net/rpc"

func main(){
    client,err := rpc.Dial("tcp","localhost:8000")
    if err!=nil{
        panic("连接失败")
    }
    var reply string
    err = client.Call("HelloService.Hello","111",&reply)
    if err!=nil{
        println(err)
    }
    println(reply)
}

tcp协议调用

go服务端

package main

import (
    "net"
    "net/rpc"
    "net/rpc/jsonrpc"
)

type HelloService struct {}

func (s *HelloService) Hello(arg string,reply *string)error{
    *reply =  "hello" + arg
    return nil
}

func main()  {
    listener,_ := net.Listen("tcp",":8000")
    _ = rpc.RegisterName("HelloService",&HelloService{})
    for{
        conn,_ := listener.Accept()
        rpc.ServeCodec(jsonrpc.NewServerCodec(conn))//替换成jsonrpc
    }
}

go客服端

package main

import (
    "net"
    "net/rpc"
    "net/rpc/jsonrpc"
)

func main(){
    conn,err := net.Dial("tcp","localhost:8000")
    if err!=nil{
        panic("连接失败")
    }
    var reply string
    client := rpc.NewClientWithCodec(jsonrpc.NewClientCodec(conn))
    err = client.Call("HelloService.Hello",",world",&reply)
    if err!=nil{
        println(err)
    }
    println(reply)
}

python客户端

import socket,json

request = {
    "id":0,
    "params":["aaaaa"],
    "method":"HelloService.Hello"
}
client = socket.create_connection(("127.0.0.1",8000))
client.sendall(json.dumps(request).encode())
rsp = client.recv(1024)
rsp = json.loads(rsp.decode())
print(rsp)
# {'id': 0, 'result': 'helloaaaaa', 'error': None}

http协议调用

go服务端

package main

import (
    "io"
    "net/http"
    "net/rpc"
    "net/rpc/jsonrpc"
)

type HelloService struct {}

func (s *HelloService) Hello(arg string,reply *string)error{
    *reply =  "hello" + arg
    return nil
}

func main()  {
    _ = rpc.RegisterName("HelloService",&HelloService{})
    http.HandleFunc("/jsonrpc",func(w http.ResponseWriter,r *http.Request){
        var conn io.ReadWriteCloser = struct {
            io.Writer
            io.ReadCloser
        }{
            ReadCloser:r.Body,
            Writer:w,
        }
        rpc.ServeRequest(jsonrpc.NewServerCodec(conn))
    })
    http.ListenAndServe(":8000",nil)
}

python客户端

import requests,json

request = {
    "id":0,
    "params":["aaaaa"],
    "method":"HelloService.Hello"
}
rsp = requests.post("http://127.0.0.1:8000/jsonrpc",json=request)
print(rsp.text)
# {'id': 0, 'result': 'helloaaaaa', 'error': None}

改良版

handler

package handler
//定义服务名
const Service = "handler/HelloService"
//定义服务对象及方法
type HelloService struct {}

func (s *HelloService) Hello(arg string,reply *string)error{
    *reply =  "hello" + arg
    return nil
}

服务端代理

package server_proxy

import (
    "awesomeProject1/msg/rpc/handler"
    "net/rpc"
)

type HelloService interface {
    Hello(arg string,reply *string)error
}
//接受接口
func RegisterHelloService(srv HelloService) error{
    return rpc.RegisterName(handler.Service,srv)
}

服务端

package main

import (
    "awesomeProject1/msg/rpc/handler"
    "awesomeProject1/msg/rpc/server/server_proxy"
    "net"
    "net/rpc"
)

func main()  {
    listener,_ := net.Listen("tcp",":8000")
    server_proxy.RegisterHelloService(&handler.HelloService{})
    for{
        conn,_ := listener.Accept()
        rpc.ServeConn(conn)
    }
}

客户端代理

package client_proxy

import (
    "awesomeProject1/msg/rpc/handler"
    "fmt"
    "net/rpc"
)

type HelloServiceStub struct {
    *rpc.Client
}

func NewHelloServiceClient(protcal,address string) HelloServiceStub{
    conn,err := rpc.Dial(protcal,address)
    if err != nil{
        panic("connect error")
    }
    return HelloServiceStub{conn}
}
func (c *HelloServiceStub) Hello(request string,reply *string) error{
    err := c.Call(handler.Service+".Hello",request,reply)
    if err != nil{
        panic("调用失败")
    }
    fmt.Println(reply)
    return err
}

客户端

package main

import (
    "awesomeProject1/msg/rpc/client/client_proxy"
    "fmt"
)

func main(){
    var reply string
    client := client_proxy.NewHelloServiceClient("tcp", "localhost:8000")
    client.Hello("aaa",&reply)
    fmt.Println(reply)
}

grpc

python

安装

pip install grpcio
pip install grpcio-tools

创建.proto文件

syntax = "proto3";

message DomoPushRequest {// 定义请求数据
    string digest = 1;
    string content= 2;
    string psd = 3;
}

message DomoPushRst{// 定义返回格式
    string rst_string = 1; //统一返回json字符串作处理
}

service DomoPushHandler {
    rpc create_domopush (DomoPushRequest) returns (DomoPushRst) {}
}

执行命令,将会生成两个py文件

python3 -m grpc_tools.protoc --python_out=. --grpc_python_out=. -I. *.proto

定义server

from concurrent import futures

import grpc

from rpc.test_pb2 import DomoPushRequest,DomoPushRst
from rpc.test_pb2_grpc import DomoPushHandlerServicer,add_DomoPushHandlerServicer_to_server

class Hello(DomoPushHandlerServicer):
    def create_domopush(self, request, context):
        print(request)
        return DomoPushRst(rst_string=request.digest+request.content+request.psd)

# 实例化server
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
# 注册
add_DomoPushHandlerServicer_to_server(Hello(),server)
# 绑定ip端口
server.add_insecure_port("[::]:50000")
server.start()
server.wait_for_termination()

定义client

import grpc

from rpc.test_pb2 import DomoPushRequest,DomoPushRst
from rpc.test_pb2_grpc import DomoPushHandlerStub

with grpc.insecure_channel("localhost:50000") as chanel:
    stub = DomoPushHandlerStub(chanel)
    response = stub.create_domopush(DomoPushRequest(digest="hello",content=",",psd="world"))
    print(response.rst_string)

go

1）安装protoc

https://github.com/protocolbuffers/protobuf/releases

2）安装protoc-gen-go

go get github.com/golang/protobuf/protoc-gen-go

3）创建proto

syntax="proto3";
option go_package = "./;proto";
service Greeter{
    rpc SayHello(HelloRequest) returns (HelloReply);
}

message HelloRequest{
    string name = 1;
}

message HelloReply{
    string message = 1;
}

4）生成go文件

protoc -I . helloworld.proto --go_out=plugins=grpc:.

5）安装grpc

go get -u google.golang.org/grpc

或下载

https://github.com/grpc/grpc-go

6）服务端

package main

//noinspection ALL
import (
    "context"
    "google.golang.org/grpc"
    "gotest/proto"
    "net"
)

type Server struct {}

func (s *Server) SayHello(ctx context.Context,request *proto.HelloRequest)(*proto.HelloReply,error){
    return &proto.HelloReply{
        Message: "hello,"+request.Name,
    },nil
}

func main(){
    g := grpc.NewServer()
    proto.RegisterGreeterServer(g,&Server{})
    lis,err := net.Listen("tcp","0.0.0.0:8080")
    if err != nil{
        panic("failed to listen"+err.Error())
    }
    g.Serve(lis)
}

7）客户端

package main
//noinspection ALL
import (
    "context"
    "fmt"
    "google.golang.org/grpc"
    "gotest/proto"
)

func main(){
    conn,err := grpc.Dial("127.0.0.1:8080",grpc.WithInsecure())
    if err != nil {
        panic(err)
    }
    defer conn.Close()
    c := proto.NewGreeterClient(conn)
    r,err := c.SayHello(context.Background(),&proto.HelloRequest{Name:"world"})
    if err != nil {
        panic(err)
    }
    fmt.Println(r.Message)
}

grpc的四种数据流

单项数据
即客户端发送一个请求给服务端，从服务端获取一个应答，就像一次普通的函数调用。
服务端数据流
客户端发起一次请求，服务端不断向客户端发送数据
客户端数据流
客户端不断向服务端发送数据，如物联网应用
双向数据流
双方相互大量发送数据，如聊天机器人

1）定义proto

syntax = "proto3";
option go_package="./;proto";
service Greeter{
    rpc GetString(StreamReqData)returns(stream StreamResData);
    rpc PutString(stream StreamReqData)returns(stream StreamResData);
    rpc AllString(stream StreamReqData)returns(stream StreamResData);
}
message  StreamReqData{
    string data = 1;
}
message  StreamResData{
    string data = 1;
}

2）服务端

package main

import (
    "fmt"
    "google.golang.org/grpc"
    "gotest/strem_test/proto"
    "net"
    "sync"
    "time"
)

type server struct {

}

//服务端流模式
func (s *server) GetString(req *proto.StreamReqData, res proto.Greeter_GetStringServer) error{
    i := 0
    fmt.Println(req.Data)
    for{
        i++
        res.Send(&proto.StreamResData{Data: fmt.Sprintf("%v",time.Now().Unix())})
        if i>10{
            break
        }
    }
    return nil
}
//客户端流模式
func (s *server) PutString(clistr proto.Greeter_PutStringServer) error{
    for{
        if a,err := clistr.Recv();err != nil{
            fmt.Println(err)
            break
        }else {
            fmt.Println(a.Data)
        }
    }
    return nil
}
//双向流模式
func (s *server) AllString(allstr proto.Greeter_AllStringServer) error{
    wg := sync.WaitGroup{}
    wg.Add(2)
    go func() {
        defer wg.Done()
        for{
            data,_ := allstr.Recv()
            fmt.Println("收到："+data.Data)
        }
    }()
    go func() {
        defer wg.Done()
        for{
            _ = allstr.Send(&proto.StreamResData{
                Data: "服务器消息",
            })
            time.Sleep(time.Second)
        }
    }()
    wg.Wait()
    return nil
}

func main()  {
    lis,err := net.Listen("tcp",":5555")
    if err != nil{
        panic(err)
    }
    s := grpc.NewServer()
    proto.RegisterGreeterServer(s,&server{})
    err = s.Serve(lis)
    if err != nil{
        panic(err)
    }

}

3）客户端

package main

import (
    "context"
    "fmt"
    "google.golang.org/grpc"
    "gotest/strem_test/proto"
    "sync"
    "time"
)

func main(){
    conn,err := grpc.Dial("localhost:5555",grpc.WithInsecure())
    if err != nil{
        panic(err)
    }
    //defer conn.Close()
    c := proto.NewGreeterClient(conn)

    /*---------------------------------*/
    res,_ := c.GetString(context.Background(),&proto.StreamReqData{Data: "hello"})
    for{
        a,err := res.Recv()
        if err != nil{
            fmt.Println(err)
            break
        }
        fmt.Println(a)
    }
    /*---------------------------------*/

    puts,_ := c.PutString(context.Background())
    i := 0
    for{
        i++
        puts.Send(&proto.StreamReqData{Data: "a"})
        time.Sleep(time.Second)
        if i>10{
            break
        }
    }

    /*---------------------------------*/
    allstr,_ := c.AllString(context.Background())

    wg := sync.WaitGroup{}
    wg.Add(2)
    go func() {
        defer wg.Done()
        for{
            data,_ := allstr.Recv()
            fmt.Println("收到："+data.Data)
        }
    }()
    go func() {
        defer wg.Done()
        for{
            _ = allstr.Send(&proto.StreamReqData{Data: "客户端消息"})
            time.Sleep(time.Second)
        }
    }()
    wg.Wait()

}

还没有评论，来说两句吧