gRPC

syntax = "proto3";
// protobuf 有2个版本，默认版本是 proto2，目前主流的版本是 proto3，因为更加易用，
// 在非空非注释第一行使用 syntax = "proto3"; 标明版本
message SearchRequest {
  // 定义 SearchRequest 的字段
  string name = 1;
  int32 number = 2;
  enum Gender {
    FEMALE = 0;
    MALE = 1;
  }
  Gender gender = 3;
  repeated string flag = 4;
}
​
message SearchResponse {
  // 定义 SearchResponse 的字段
  int32 code = 1;
  string msg = 2;
  repeated string data = 3;
}
​
// 定义 SearchService 的 RPC 服务接口Search
service SearchService {
  rpc Search(SearchRequest) returns (SearchResponse);
}

import "google/protobuf/any.proto";
message ErrorStatus {
  string message = 1;
  repeated google.protobuf.Any details = 2;
}
// 表示details列表中元素不指定具体类型

oneof oneof_name {
  field_type1 field_name1 = field_number1;
  field_type2 field_name2 = field_number2;
}

syntax = "proto3";
message Person {
  string name = 1;
  int32 age = 2;
  oneof info {
    string email = 3;
    string phone = 4;
  }
}

syntax = "proto3";
message UserMessage {
  map<string, int32> dictionary = 1;
}

# 添加键值对
test_pb2.UserMessage.dictionary["num"] = 23
test_pb2.UserMessage.dictionary["age"] = 20
# 获取键值对
test_pb2.UserMessage.dictionary["num"]
test_pb2.UserMessage.dictionary.get("age")

pip install grpcio
pip install grpcio-tools

python -m grpc_tools.protoc --python_out=.  --grpc_python_out=.  -I. test.proto

import grpc
from concurrent import futures
from protoc_utils import test_pb2
from protoc_utils import test_pb2_grpc
import time
​
class SearchService(test_pb2_grpc.SearchServiceServicer):
    def Search(self, request, context):
        # 获取请求元数据
        metadata = context.invocation_metadata()
        print(metadata)
        # 发送自定义元数据给客户端
        custom_metadata = [('key1', 'value1'), ('key2', 'value2')]
        context.send_initial_metadata(custom_metadata)
        # 获取客户端的 IP 地址
        client_ip = context.peer()
        print(client_ip)
        # 获取请求的剩余时间
        remaining_time = context.time_remaining()
        print(remaining_time)
        # 检查请求是否已被取消
        # time.sleep(10)
        if context.is_active():
            # 处理请求...
            print(111)
        else:
            print(222)
        # 处理 Search 请求
        name = request.name
        number = request.number
        gender = request.gender
        flags = request.flag
        print(name, number, gender, flags, type(flags))
        # 执行搜索逻辑
        pass
        # 创建并返回 SearchResponse
        response = test_pb2.SearchResponse()
        response.code = 200
        response.msg = "Search completed successfully."
        # data是重复列表字段需要用extend赋值
        response.data.extend(["test", "234", 'ok'])
        return response
​
def serve():
    # 创建 gRPC 服务器，这里可定义最大接收和发送大小(单位M)，默认只有4M
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10), options=[
        ('grpc.max_send_message_length', 100 * 1024 * 1024),
        ('grpc.max_receive_message_length', 100 * 1024 * 1024)])
    # server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    # 注册 SearchService
    test_pb2_grpc.add_SearchServiceServicer_to_server(SearchService(), server)
    # 监听端口并启动服务
    server.add_insecure_port('[::]:50051')
    server.start()
    print("Server started. Listening on port 50051.")
    # 阻塞主线程，直到服务器关闭
    try:
        while True:
            pass
    except KeyboardInterrupt:
        server.stop(0)
if __name__ == '__main__':
    serve()

import grpc
from protoc_utils import test_pb2
from protoc_utils import test_pb2_grpc

def run():
    # 创建 gRPC 通道
    channel = grpc.insecure_channel('localhost:50051')
    # 创建 SearchService 的客户端
    stub = test_pb2_grpc.SearchServiceStub(channel)
    # 创建 SearchRequest 对象
    request = test_pb2.SearchRequest()
    request.name = "bob"
    request.number = 10
    # 枚举可直接拿到proto中键对应的值进行赋值
    request.gender = test_pb2.SearchRequest.MALE
    # flag是重复列表字段需要用extend赋值
    request.flag.extend(["ok", "123", "222"])
    # 调用 Search 方法
    response = stub.Search(request)
    print(response, type(response))
    # 处理 SearchResponse
    print("Code:", response.code)
    print("Message:", response.msg, type(response.msg))
    print("Data:", response.data, type(response.data))
    l = list(response.data)
    print("Data:", l, type(l))
if __name__ == '__main__':
    run()