[网络协议]grpc-python 错误码与服务端客户端发送与接收错误信息

错误码与服务端客户端发送与接收错误信息

失败的code码表格

代码示例

1. helloworld.proto:

syntax = "proto3";  //指定版本

package test;     // 包的命名名称

service grpc_test {
	// 定义一个rpc函数
	rpc
			test_client(stream test_client_requests)
			returns(test_client_response){}
}

message test_client_requests{
	string data = 1;
}

message test_client_response{
	string result = 1;
}

2. server.py:

import time
import grpc
import helloworld_pb2 as pb2
import helloworld_pb2_grpc as pb2_grpc


from concurrent import futures   # 进程线程包


class grpc_test(pb2_grpc.grpc_testServicer):

    def test_client(self, request_iterator, context):
        # context.is_active()     # 检测客户端是否还活着

        # 接收来自客户端流数据
        index = 0
        for request in request_iterator:
            data = request.data
            index += 1
            result = f'receive {data},index:{index}'
            print(result)
            time.sleep(1)
            # 假设不符合条件抛出错误码
            if index == 2:
                import json
                response_data = {"sub_code": 40301, "message": 'xxx错误！'}
                context.set_details(json.dumps(response_data))  # 设置报错信息
                context.set_code(grpc.StatusCode.DATA_LOSS)     # 指定状态码错误类型
                raise context
                # context.cancel()        # 强制终止连接，会引起客户端抛出异常
        return pb2.test_client_response(result="ok")


def run():
    grpc_server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=4)   # 指定线程数
    )
    pb2_grpc.add_grpc_testServicer_to_server(grpc_test(), grpc_server)  # 注册服务到grpc_server里
    grpc_server.add_insecure_port('0.0.0.0:50010')                       # 指定ip、端口号
    print('server will start at 0.0.0.0:50010')
    grpc_server.start()

    # 防止启动后直接结束添加阻塞，在其他语言不会出现这样的问题
    try:
        while 1:
            time.sleep(3600)
    except Exception as e:
        grpc_server.stop(0)
        print(e)

if __name__ == '__main__':
    run()

3. client.py:

import json
import random
import time

import grpc
import helloworld_pb2 as pb2
import helloworld_pb2_grpc as pb2_grpc


def test():
    """
    循环发送流数据
    每一秒中向服务端发送流数据
    :return:
    """
    index = 0
    while True:
        index += 1
        time.sleep(1)
        str_random = str(random.random())
        yield pb2.test_client_requests(
            data=str_random
        )
        # 条件终止连接
        if index == 15:
            break


def run():
    # 定义一个频道 绑定ip、端口号
    conn = grpc.insecure_channel('127.0.0.1:50010')

    # 生成客户端
    client = pb2_grpc.grpc_testStub(channel=conn)

    # 传入参数获取返回值
    try:
        response = client.test_client(test(), timeout=15)       # timeout超出时长抛出异常

        print(response)         # 正常的返回内容
    except Exception as e:
        print(dir(e))                   # 返回模块的属性列表。
        code_name = e.code().name       # 错误类型
        code_value = e.code().value     # 错误状态码
        details = e.details()           # 错误信息
        print(code_name)
        print(code_value)
        print(json.loads(details))


if __name__ == '__main__':
    run()

结果图: