脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 脚本专栏 → python → Python grpc服务

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文件头部声明#coding=utf-8问题

    python文件头部声明#coding=utf-8问题

    这篇文章主要介绍了python文件头部声明#coding=utf-8问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2024-06-06
  • Python数据集切分实例

    Python数据集切分实例

    今天小编就为大家分享一篇Python数据集切分实例，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
    2018-12-12
  • python多线程同步实例教程

    python多线程同步实例教程

    这篇文章主要给大家介绍了关于python多线程同步的相关资料，文中通过示例代码介绍的非常详细，对大家学习或者使用python具有一定的参考学习价值，需要的朋友们下面来一起学习学习吧
    2019-08-08
  • python导入SciPy的io模块使用

    python导入SciPy的io模块使用

    SciPy是一个强大的科学计算库,本文就就来介绍了python导入SciPy的io模块使用,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2025-07-07
  • 关于pyqtSignal的基本使用

    关于pyqtSignal的基本使用

    这篇文章主要介绍了关于pyqtSignal的基本使用方式，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2022-06-06
  • Python 打印中文字符的三种方法

    Python 打印中文字符的三种方法

    本文给大家分享三种方法实现python打印中文字符的方法，代码简单易懂，非常不错，具有一定的参考借鉴价值，需要的朋友参考下吧
    2018-08-08
  • 编译 pycaffe时报错：fatal error: numpy/arrayobject.h没有那个文件或目录

    编译 pycaffe时报错：fatal error: numpy/arrayobject.h没有那个文件或目录

    这篇文章主要介绍了编译 pycaffe时报错：fatal error: numpy/arrayobject.h没有那个文件或目录,需要的朋友可以参考下
    2020-11-11
  • python爬取2021猫眼票房字体加密实例

    python爬取2021猫眼票房字体加密实例

    在本篇文章里小编给大家整理的是一篇关于python爬取2021猫眼票房字体加密实例内容，有兴趣的朋友们可以学习下。
    2021-02-02
  • 使用Django Form解决表单数据无法动态刷新的两种方法

    使用Django Form解决表单数据无法动态刷新的两种方法

    这篇文章主要介绍了使用Django Form解决表单数据无法动态刷新的两种方法,需要的朋友可以参考下
    2017-07-07
  • Python读取word文本操作详解

    Python读取word文本操作详解

    这篇文章主要介绍了Python读取word文本操作详解，介绍了涉及到的模块，相关概念，模块的安装等内容，小编觉得还是挺不错的，具有一定借鉴价值，需要的朋友可以参考下
    2018-01-01

最新评论