Python使用grpc服務(wù)并與C++互相調(diào)用

更新時(shí)間：2025年12月15日 09:13:29 作者：MC皮蛋俠客

這篇文章主要為大家詳細(xì)介紹了gRPC的安裝與使用,包括使用pip安裝相關(guān)組件,并和C++互相調(diào)用gRPC服務(wù),文中的示例代碼講解詳細(xì),感興趣的小伙伴可以了解下

安裝grpc

使用pip安裝 gRPC 和 Protocol Buffers (protobuf) 編譯器（最好在虛擬環(huán)境下安裝，防止干擾系統(tǒng)環(huán)境）

pip install grpcio grpcio-tools

使用grpc

1.使用 protoc（Protocol Buffers 編譯器）和 gRPC 插件來(lái)從 .proto 文件生成 Python 代碼。

device.proto文件內(nèi)容

syntax = "proto3";

package device_service;

service DeviceService {
  // 返回一維字符串?dāng)?shù)組
  rpc GetDeviceStringList (DeviceNameListRequest) returns (DeviceNameListResponse) {}
  // 返回單個(gè)int值
  rpc GetDeviceSlaveCnt (DeviceSlaveCntRequest) returns (DeviceSlaveCntResponse) {}
  // 返回自定義結(jié)構(gòu)體類(lèi)型
  rpc GetDeviceInfo (DeviceInfoRequest) returns (DeviceInfoResponse) {}
  // 返回二維字符串?dāng)?shù)組
  rpc GetDeviceTableBySlaveId (DeviceTableBySlaveIdRequest) returns (DeviceTableBySlaveIdResponse) {}
}

// 設(shè)備列表信息
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;
}

// 設(shè)備詳細(xì)信息
message DeviceInfoRequest {
  // 查詢(xún)的設(shè)備名稱(chēng)
  string device_name = 1;
}

message DeviceInfoResponse {
  DeviceInfoDetail info = 1;
}

// 獲取設(shè)備從機(jī)數(shù)量
message DeviceSlaveCntRequest {
  string device_name = 1;
}

message DeviceSlaveCntResponse {
  int32 slave_cnt = 1;
}

// 根據(jù)設(shè)備名稱(chēng)和從機(jī)id和測(cè)點(diǎn)信息 獲取設(shè)備詳細(xì)信息,回復(fù)是二維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;
}

在命令行中運(yùn)行以下命令：

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

2.編寫(xiě)服務(wù)端和客戶(hù)端代碼

編寫(xiě)服務(wù)端代碼

1.檢查設(shè)備是否存在

def checkDevice(self, request, context) -> bool:
    device = self.device_map.get(request.device_name)
    if device is None:
        # 處理設(shè)備不存在的情況，例如返回一個(gè)錯(cuò)誤消息或拋出一個(gè)自定義異常
        context.set_code(grpc.StatusCode.NOT_FOUND)
        context.set_details(f'未找到{request.device_name}設(shè)備')
        print(f'未找到{request.device_name}設(shè)備')
        return False
    else:
        return True

2.返回一維字符串列表

proto文件內(nèi)容

service DeviceService {
  // 返回一維字符串?dāng)?shù)組
  rpc GetDeviceStringList (DeviceNameListRequest) returns (DeviceNameListResponse) {}
}

// 設(shè)備列表信息
message DeviceNameListRequest {
  string device_name = 1;
}

message DeviceNameListResponse {
  repeated string device_names = 1;
}

python服務(wù)端代碼

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

3.返回單個(gè)int類(lèi)型

proto文件內(nèi)容

service DeviceService {
  // 返回單個(gè)int值
  rpc GetDeviceSlaveCnt (DeviceSlaveCntRequest) returns (DeviceSlaveCntResponse) {}
}

// 獲取設(shè)備從機(jī)數(shù)量
message DeviceSlaveCntRequest {
  string device_name = 1;
}

message DeviceSlaveCntResponse {
  int32 slave_cnt = 1;
}

python服務(wù)端代碼

def GetDeviceSlaveCnt(self, request, context):
    if not self.checkDevice(request, context):
        return device_pb2.DeviceSlaveCntResponse(slave_cnt=0)
    try:
        # 創(chuàng)建并返回響應(yīng)
        response = device_pb2.DeviceSlaveCntResponse(slave_cnt=1)
        return response
    except Exception as e:
        print(e)
        return device_pb2.DeviceSlaveCntResponse(slave_cnt=0)

4.返回自定義結(jié)構(gòu)體類(lèi)型

proto文件內(nèi)容

service DeviceService {
  // 返回自定義結(jié)構(gòu)體類(lèi)型
  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;
}

// 設(shè)備詳細(xì)信息
message DeviceInfoRequest {
  // 查詢(xún)的設(shè)備名稱(chēng)
  string device_name = 1;
}

// 返回值是自定義結(jié)構(gòu)體類(lèi)型DeviceInfo
message DeviceInfoResponse {
  DeviceInfoDetail info = 1;
}

python服務(wù)端代碼

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
        )
        # 創(chuàng)建并返回響應(yīng)
        response = device_pb2.DeviceInfoResponse(info=info_detail)
        return response
    except Exception as e:
        print(e)
        return device_pb2.DeviceInfoResponse()

5.返回二維字符串列表

proto文件內(nèi)容

service DeviceService {
  // 返回二維字符串?dāng)?shù)組
  rpc GetDeviceTableBySlaveId (DeviceTableBySlaveIdRequest) returns (DeviceTableBySlaveIdResponse) {}
}

// 根據(jù)設(shè)備名稱(chēng)和從機(jī)id和測(cè)點(diǎn)信息 獲取設(shè)備詳細(xì)信息,回復(fù)是二維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服務(wù)端代碼

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:
        # 創(chuàng)建并返回響應(yīng)
        head_data = ["設(shè)備名稱(chēng)","設(shè)備類(lèi)型","設(shè)備IP","設(shè)備端口","設(shè)備狀態(tài)","模擬狀態(tài)","計(jì)劃狀態(tài)"]
        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.啟動(dòng)服務(wù)端

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

if __name__ == '__main__':
    serve()

編寫(xiě)客戶(hù)端代碼

初始化客戶(hù)端類(lèi)

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

獲取單個(gè)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

獲取自定義結(jié)構(gòu)體信息

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]
    # 處理響應(yīng)中的二維數(shù)組數(shù)據(jù)
    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

關(guān)閉通道

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

啟動(dòng)客戶(hù)端

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服務(wù)端和客戶(hù)端通信效果展示

Python和C++互相調(diào)用grpc服務(wù)

項(xiàng)目源碼倉(cāng)庫(kù)地址https://gitee.com/chen-dongyu123/grpc_example

1.python和c++需要共同使用同一份.proto文件，生產(chǎn)各自的grpc服務(wù)端和客戶(hù)端代碼

2.根據(jù)需求確定python和c++誰(shuí)當(dāng)客戶(hù)端和服務(wù)端，編寫(xiě)完成各自的服務(wù)端和客戶(hù)端代碼

這里是我的一個(gè)例子

python當(dāng)服務(wù)端，c++當(dāng)客戶(hù)端(c++使用grpc的例子可以看我上一篇文章)

c++當(dāng)服務(wù)端，python當(dāng)客戶(hù)端

總結(jié)

python中使用grpc就比在c++簡(jiǎn)單多了，我們可以使用grpc輕松的跨語(yǔ)言通訊，相比傳統(tǒng)的webserver通訊，grpc的效率更高。對(duì)于追求效率的場(chǎng)景下，我們可以使用c++編寫(xiě)，然后業(yè)務(wù)方面我們可以使用Java或者python。我現(xiàn)在遇到的一個(gè)場(chǎng)景就是需要使用modbus實(shí)時(shí)采集數(shù)據(jù)，然后將數(shù)據(jù)傳遞給web后臺(tái)或者客戶(hù)端，對(duì)于這種場(chǎng)景，grpc就比webserver優(yōu)勢(shì)大多了。

以上就是Python使用grpc服務(wù)并與C++互相調(diào)用的詳細(xì)內(nèi)容，更多關(guān)于Python grpc服務(wù)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: