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

更新时间：2025年12月15日 09:13:29 作者：MC皮蛋侠客

这篇文章主要为大家详细介绍了gRPC的安装与使用,包括使用pip安装相关组件,并和C++互相调用gRPC服务,文中的示例代码讲解详细,感兴趣的小伙伴可以了解下

安装grpc

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

pip install grpcio grpcio-tools

使用grpc

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服务端和客户端通信效果展示

Python和C++互相调用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

Python3.4学习笔记之常用操作符,条件分支和循环用法示例
这篇文章主要介绍了Python3.4常用操作符,条件分支和循环用法,结合实例形式较为详细的分析了Python3.4常见的数学运算、逻辑运算操作符,条件分支语句,循环语句等功能与基本用法,需要的朋友可以参考下
2019-03-03
解决Python import .pyd 可能遇到路径的问题
这篇文章主要介绍了解决Python import .pyd 可能遇到路径的问题，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2021-03-03
python Graham求凸包问题并画图操作
这篇文章主要介绍了python Graham求凸包问题并画图操作，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-06-06
Python机器学习应用之支持向量机的分类预测篇
最近完成的一个项目用到了SVM，之前也一直有听说支持向量机，知道它是机器学习中一种非常厉害的算法。利用将近一个星期的时间学习了一下支持向量机，把原理推了一遍，感觉支持向量机确实挺厉害的，这篇文章带你了解它
2022-01-01
python pycurl验证basic和digest认证的方法
这篇文章主要介绍了python pycurl验证basic和digest认证的方法，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2018-05-05
Python实现Excel做表自动化的最全方法合集
Microsoft Excel 是一款强大的办公工具,广泛用于数据分析、报告制作、预算管理等各种任务,本文将深入探讨如何使用 Python 进行 Excel 表格的自动化,需要的可以参考下
2024-02-02
浅析Python中return和finally共同挖的坑
最近在工作中遇到一个坑，发现这个坑居然存在于return和finally，所以觉着有必要总结分享一下，下面这篇文章主要介绍了关于Python中return和finally共同挖的坑，需要的朋友可以参考借鉴，下面来一起看看吧。
2017-08-08
Python实现敏感词过滤的五种方法
在我们生活中的一些场合经常会有一些不该出现的敏感词,我们通常会使用*去屏蔽它,一些骂人的敏感词和一些政治敏感词都不应该出现在一些公共场合中,这个时候我们就需要一定的手段去屏蔽这些敏感词,下面我来介绍一些简单版本的Python敏感词屏蔽的方法,需要的朋友可以参考下
2025-04-04
使用Pytest.main()运行时参数不生效问题解决
本文主要介绍了使用Pytest.main()运行时参数不生效问题解决，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2023-02-02
Python cookbook（数据结构与算法）将序列分解为单独变量的方法
这篇文章主要介绍了Python cookbook（数据结构与算法）将序列分解为单独变量的方法,结合实例形式分析了Python序列赋值实现的分解成单独变量功能相关操作技巧,需要的朋友可以参考下
2018-02-02

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

目录

安装grpc

使用grpc

Python和C++互相调用grpc服务

总结

相关文章

最新评论

大家感兴趣的内容

最近更新的内容

常用在线小工具