python使用grpc

news2024/11/6 17:29:29

proto

首先编写proto，也可以根据对象生成proto

syntax = "proto3";


package text;

service TextSender{
    rpc Send(Text) returns (SendResponse);

    rpc Resend(Text) returns (SendResponse);
}

message Text{
    string text = 1;
}

message SendResponse{
    bool sucess = 1;
    string para = 2;
}

生成py文件

py -m grpc_tools.protoc -I/ --python_out=. --pyi_out=. --grpc_python_out=. first.proto

编写

根据proto生成的文件编写函数

# server.py
from concurrent import futures
import logging

import grpc
import first_pb2
import first_pb2_grpc

class Sender(first_pb2_grpc.TextSenderServicer):
    def Send(self, request, context):
        return first_pb2.SendResponse(sucess=True, para="first")
    
    def Resend(self, request, context):
        return first_pb2.SendResponse(sucess=True, para="second")
    
def serve():
    port = "50051"
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    first_pb2_grpc.add_TextSenderServicer_to_server(Sender(), server)
    server.add_insecure_port("[::]:" + port)
    server.start()
    print("Server started, listening on " + port)
    server.wait_for_termination()


if __name__ == "__main__":
    logging.basicConfig()
    serve()

# client.py
from __future__ import print_function

import logging

import grpc
import first_pb2
import first_pb2_grpc


def run():
    print("Will try to greet world ...")
    with grpc.insecure_channel("39.105.170.229:50051") as channel:
        stub = first_pb2_grpc.TextSenderStub(channel)
        response = stub.Send(first_pb2.Text(text="you"))
        print(f"Greeter client received: {response.sucess} for the {response.para} times")
        response = stub.Resend(first_pb2.Text(text='you'))
        print(f"Greeter client received: {response.sucess} for the {response.para} times")

if __name__ == "__main__":
    logging.basicConfig()
    run()

启动

在这里插入图片描述

注册到nacos

import nacos
import time

SERVER_ADDRESSES = "http://1.2.3.4:8848"  # Nacos服务器地址
NAMESPACE = "d344b685-a423-4770-a6e4-da81245b5dc9"  # Nacos的命名空间ID

# 获取Nacos客户端
client = nacos.NacosClient(SERVER_ADDRESSES, namespace=NAMESPACE, username="nacos", password="nacos")

# 服务注册
client.add_naming_instance("grpc-demo", "1.2.3.4", port=50051)


while True:
    try:
        client.send_heartbeat("grpc-demo", "1.2.3.4", port=50051)
        time.sleep(30)
    except Exception as e:
        print(f"Error: {e}")
        time.sleep(5)  # 在尝试重新发送心跳之前稍作延迟