Python使用grpc服务并与C++互相调用

目录

• 安装grpc
• 使用grpc
• Python和C++互相调用grpc服务
• 总结

使用pip安装 gRPC 和 Protocol Buffers (protobuf) 编译器（最好在虚拟环境下安装，防止干扰系统环境）

pip install grpcio grpcio-tools

1.使用 protoc（Protocol Buffers 编译器）和 gRPC 插件来从 .proto 文件生成 Python 代码。

device.proto文件内容

syntax = "proto3";

package device_service;

service DeviceService {
  // 返回一维字符串数组
  rpc GetDeviceStringList (DeviceNameListRequest) returns (DeviceNameListResponse) {}
  // 返回单个int值
  rpc GetDeviceSlaveCnt (DeviceSlaveCntRequest) returns (DeviceSlaveCntResponse) {}
  // 返回自定义结构体类型
  rpc GetDeviceInfo (DeviceInfoRequest) returns (DeviceInfoResponse) {}
  // 返回二维字符串数组
  rpc GetDeviceTableBySlaveId (DeviceTableBySlaveIdRequest) returns (DeviceTableBySlaveIdResponse) {}
}

// 设备列表信息
message DeviceNameListRequest {
  string device_name = 1;
}

message DeviceNameListResponse {
  repeated string device_names = 1;
}

message DeviceInfoDetail {
  string ip = 1;
  int32 port = 2;
  string type = 3;
  bool server_status = 4;
  bool simulate_status = 5;
  bool plan_status = 6;
}

// 设备详细信息
message DeviceInfoRequest {
  // 查询的设备名称
  string device_name = 1;
}

message DeviceInfoResponse {
  DeviceInfoDetail info = 1;
}

// 获取设备从机数量
message DeviceSlaveCntRequest {
  string device_name = 1;
}

message DeviceSlaveCntResponse {
  int32 slave_cnt = 1;
}

// 根据设备名称和从机id和测点信息 获取设备详细信息,回复是二维List
message DeviceTableBySlaveIdRequest {
  string device_name = 1;
  int32 slave_id = 2;
  string point_name = 3;
}

message DeviceTableRow{
  repeated string row = 1;
}

message DeviceTableBySlaveIdResponse {
  DeviceTableRow head_data = 1;
  repeated DeviceTableRow table_data = 2;
}

在命令行中运行以下命令：

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

2.编写服务端和客户端代码

编写服务端代码

1.检查设备是否存在

def checkDevice(self, request, context) -> bool:
    device = self.device_map.get(request.device_name)
    if device is None:
        # 处理设备不存在的情况，例如返回一个错误消息或抛出一个自定义异常
        context.set_code(grpc.StatusCode.NOT_FOUND)
        context.set_details(f'未找到{request.device_name}设备')
        print(f'未找到{request.device_name}设备')
        return False
    else:
        return True

2.返回一维字符串列表

proto文件内容

service DeviceService {
  // 返回一维字符串数组
  rpc GetDeviceStringList (DeviceNameListRequest) returns (DeviceNameListResponse) {}
}

// 设备列表信息
message DeviceNameListRequest {
  string device_name = 1;
}

message DeviceNameListResponse {
  repeated string device_names = 1;
}

python服务端代码

def GetDeviceStringList(self, request, context):
    device_name_list = ["PCS1", "PCS2", "BMS1", "BMS2"]
    return device_pb2.DeviceNameListResponse(device_names=device_name_list)

3.返回单个int类型

proto文件内容

service DeviceService {
  // 返回单个int值
  rpc GetDeviceSlaveCnt (DeviceSlaveCntRequest) returns (DeviceSlaveCntResponse) {}
}

// 获取设备从机数量
message DeviceSlaveCntRequest {
  string device_name = 1;
}

message DeviceSlaveCntResponse {
  int32 slave_cnt = 1;
}

python服务端代码

def GetDeviceSlaveCnt(self, request, context):
    if not self.checkDevice(request, context):
        return device_pb2.DeviceSlaveCntResponse(slave_cnt=0)
    try:
        # 创建并返回响应
        response = device_pb2.DeviceSlaveCntResponse(slave_cnt=1)
        return response
    except Exception as e:
        print(e)
        return device_pb2.DeviceSlaveCntResponse(slave_cnt=0)

4.返回自定义结构体类型

proto文件内容

service DeviceService {
  // 返回自定义结构体类型
  rpc GetDeviceInfo (DeviceInfoRequest) returns (DeviceInfoResponse) {}
}

message DeviceInfoDetail {
  string ip = 1;
  int32 port = 2;
  string type = 3;
  bool server_status = 4;
  bool simulate_status = 5;
  bool plan_status = 6;
}

// 设备详细信息
message DeviceInfoRequest {
  // 查询的设备名称
  string device_name = 1;
}

// 返回值是自定义结构体类型DeviceInfo
message DeviceInfoResponse {
  DeviceInfoDetail info = 1;
}

python服务端代码

def GetDeviceInfo(self, request, context):
    if not self.checkDevice(request, context):
        return device_pb2.DeviceInfoResponse()
    try:
        info_detail = device_pb2.DeviceInfoDetail(
            ip="127.0.0.1",
            port=502,
            type="tcp",
            server_status=True,
            simulate_status=True,
            plan_status=False
        )
        # 创建并返回响应
        response = device_pb2.DeviceInfoResponse(info=info_detail)
        return response
    except Exception as e:
        print(e)
        return device_pb2.DeviceInfoResponse()

5.返回二维字符串列表

proto文件内容

service DeviceService {
  // 返回二维字符串数组
  rpc GetDeviceTableBySlaveId (DeviceTableBySlaveIdRequest) returns (DeviceTableBySlaveIdResponse) {}
}

// 根据设备名称和从机id和测点信息 获取设备详细信息,回复是二维List
message DeviceTableBySlaveIdRequest {
  string device_name = 1;
  int32 slave_id = 2;
  string point_name = 3;
}

message DeviceTableRow{
  repeated string row = 1;
}

message DeviceTableBySlaveIdResponse {
  DeviceTableRow head_data = 1;
  repeated DeviceTableRow table_data = 2;
}

python服务端代码

def GetDeviceTableBySlaveId(self, request, context):
    if not self.checkDevice(request, context):
        return device_pb2.DeviceTableBySlaveIdResponse(head_data=device_pb2.DeviceTableRow(row=[]),
                                                        table_data=[device_pb2.DeviceTableRow(row=[])])
    try:
        # 创建并返回响应
        head_data = ["设备名称","设备类型","设备IP","设备端口","设备状态","模拟状态","计划状态"]
        table_data = [["PCS1", "tcp", "127.0.0.1", "502", "true", "true", "false"],
                        ["PCS2", "tcp", "127.0.0.1", "503", "true", "true", "false"],
                        ["BMS1", "tcp", "127.0.0.1", "504", "true", "true", "false"],
                        ["BMS2", "tcp", "127.0.0.1", "505", "true", "true", "false"]]
        # 封装成rpc格式
        head_row = device_pb2.DeviceTableRow(row=head_data)
        response = device_pb2.DeviceTableBySlaveIdResponse(
            head_data=head_row,
            table_data=[device_pb2.DeviceTableRow(row=row_data) for row_data in table_data]
        )
        return response
    except Exception as e:
        print(e)
        return device_pb2.DeviceTableBySlaveIdResponse(head_data=device_pb2.DeviceTableRow(row=[]),
                                                        table_data=[device_pb2.DeviceTableRow(row=[])])

6.启动服务端

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    device_pb2_grpc.add_DeviceServiceServicer_to_server(DeviceServiceServicer(), server)
    server.add_insecure_port('[::]:50051')	# 端口号可以自定义
    server.start()
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

 编写客户端代码

初始化客户端类

class DeviceServiceClient:
    def __init__(self, channel_address='localhost:50051'):
        self.channel = grpc.insecure_channel(channel_address)
        self.stub = device_pb2_grpc.DeviceServiceStub(self.channel)

获取一维字符串列表

def getDeviceStringList(self) -> List[str]:
    device_names_response = self.stub.GetDeviceStringList(device_pb2.DeviceNameListRequest())
    device_names = [name for name in device_names_response.device_names]
    return device_names

获取单个int值

def getDeviceSlaveCnt(self, device_name: str) -> int:
    device_slave_cnt_response = self.stub.GetDeviceSlaveCnt(
        device_pb2.DeviceSlaveCntRequest(device_name=device_name))
    slave_cnt = device_slave_cnt_response.slave_cnt
    return slave_cnt

获取自定义结构体信息

def getDeviceInfo(self, device_name: str) -> Dict:
    device_info_response = self.stub.GetDeviceInfo(device_pb2.DeviceInfoRequest(device_name=device_name))
    device_info_dict = {
        "ip": device_info_response.info.ip,
        "port": device_info_response.info.port,
        "type": device_info_response.info.type,
        "server_status": device_info_response.info.server_status,
        "simulate_status": device_info_response.info.simulate_status,
        "plan_status": device_info_response.info.plan_status
    }
    return device_info_dict

获取二维字符串列表

def getDeviceTableBySlaveId(self, device_name: str, slave_id: int, point_name: str = "") -> Dict:
    response = self.stub.GetDeviceTableBySlaveId(
        device_pb2.DeviceTableBySlaveIdRequest(device_name=device_name, slave_id=slave_id, point_name=point_name))
    head_data = [head for head in response.head_data.row]
    # 处理响应中的二维数组数据
    table_data = []
    for value in response.table_data:
        row_data = [item for item in value.row]
        table_data.append(row_data)
        data_dict = {
            "head_data": head_data,
            "table_data": table_data
        }
        return data_dict

关闭通道

def close(self):
    self.channel.close()

启动客户端

if __name__ == "__main__":
    client = DeviceServiceClient()

    device_name_list = client.getDeviceStringList()
    print(device_name_list)

    device_info_dict = client.getDeviceInfo("PCS1")
    print(device_info_dict)

    slave_cnt = client.getDeviceSlaveCnt("PCS1")
    print(slave_cnt)

    table_data_dict = client.getDeviceTableBySlaveId("PCS1", 1, "")
    print(table_data_dict)

    client.close()

grpc服务端和客户端通信效果展示

项目源码仓库地址https://gitee.com/chen-dongyu123/grpc_example

1.python和c++需要共同使用同一份.proto文件，生产各自的grpc服务端和客户端代码

2.根据需求确定python和c++谁当客户端和服务端，编写完成各自的服务端和客户端代码

这里是我的一个例子

python当服务端，c++当客户端(c++使用grpc的例子可以看我上一篇文章)

c++当服务端，python当客户端

python中使用grpc就比在c++简单多了，我们可以使用grpc轻松的跨语言通讯，相比传统的webserver通讯，grpc的效率更高。对于追求效率的场景下，我们可以使用c++编写，然后业务方面我们可以使用Java或者python。我现在遇到的一个场景就是需要使用modbus实时采集数据，然后将数据传递给web后台或者客户端，对于这种场景，grpc就比webserver优势大多了。

以上就是Python使用grpc服务并与C++互相调用的详细内容，更多关于Python grpc服务的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• python grpc实现异步调用(不用grpc异步接口)
• Python使用gRPC实现数据分析能力的共享
• 利用Python语言的grpc实现消息传送详解
• Python gRPC流式通信协议详细讲解
• 用Python编写简单的gRPC服务的详细过程
• Python grpc超时机制代码示例