Python使用grpc服務并與C++互相調(diào)用
更新時間：2025年12月15日 09:13:29 作者：MC皮蛋俠客
這篇文章主要為大家詳細介紹了gRPC的安裝與使用,包括使用pip安裝相關組件,并和C++互相調(diào)用gRPC服務,文中的示例代碼講解詳細,感興趣的小伙伴可以了解下
安裝grpc
使用pip安裝 gRPC 和 Protocol Buffers (protobuf) 編譯器（最好在虛擬環(huán)境下安裝，防止干擾系統(tǒng)環(huán)境）
pip install grpcio grpcio-tools
使用grpc
1.使用 protoc（Protocol Buffers 編譯器）和 gRPC 插件來從 .proto 文件生成 Python 代碼。
device.proto文件內(nèi)容
syntax = "proto3";
package device_service;
service DeviceService {
// 返回一維字符串數(shù)組
rpc GetDeviceStringList (DeviceNameListRequest) returns (DeviceNameListResponse) {}
// 返回單個int值
rpc GetDeviceSlaveCnt (DeviceSlaveCntRequest) returns (DeviceSlaveCntResponse) {}
// 返回自定義結構體類型
rpc GetDeviceInfo (DeviceInfoRequest) returns (DeviceInfoResponse) {}
// 返回二維字符串數(shù)組
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;
}
// 獲取設備從機數(shù)量
message DeviceSlaveCntRequest {
string device_name = 1;
}
message DeviceSlaveCntResponse {
int32 slave_cnt = 1;
}
// 根據(jù)設備名稱和從機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文件內(nèi)容
service DeviceService {
// 返回一維字符串數(shù)組
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文件內(nèi)容
service DeviceService {
// 返回單個int值
rpc GetDeviceSlaveCnt (DeviceSlaveCntRequest) returns (DeviceSlaveCntResponse) {}
}
// 獲取設備從機數(shù)量
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:
# 創(chuàng)建并返回響應
response = device_pb2.DeviceSlaveCntResponse(slave_cnt=1)
return response
except Exception as e:
print(e)
return device_pb2.DeviceSlaveCntResponse(slave_cnt=0)
4.返回自定義結構體類型
proto文件內(nèi)容
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
)
# 創(chuà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 {
// 返回二維字符串數(shù)組
rpc GetDeviceTableBySlaveId (DeviceTableBySlaveIdRequest) returns (DeviceTableBySlaveIdResponse) {}
}
// 根據(jù)設備名稱和從機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:
# 創(chuàng)建并返回響應
head_data = ["設備名稱","設備類型","設備IP","設備端口","設備狀態(tài)","模擬狀態(tài)","計劃狀態(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.啟動服務端
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]
# 處理響應中的二維數(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
關閉通道
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++互相調(diào)用grpc服務
項目源碼倉庫地址https://gitee.com/chen-dongyu123/grpc_example
1.python和c++需要共同使用同一份.proto文件，生產(chǎn)各自的grpc服務端和客戶端代碼
2.根據(jù)需求確定python和c++誰當客戶端和服務端，編寫完成各自的服務端和客戶端代碼
這里是我的一個例子
python當服務端，c++當客戶端(c++使用grpc的例子可以看我上一篇文章)
c++當服務端，python當客戶端
總結
python中使用grpc就比在c++簡單多了，我們可以使用grpc輕松的跨語言通訊，相比傳統(tǒng)的webserver通訊，grpc的效率更高。對于追求效率的場景下，我們可以使用c++編寫，然后業(yè)務方面我們可以使用Java或者python。我現(xiàn)在遇到的一個場景就是需要使用modbus實時采集數(shù)據(jù)，然后將數(shù)據(jù)傳遞給web后臺或者客戶端，對于這種場景，grpc就比webserver優(yōu)勢大多了。
以上就是Python使用grpc服務并與C++互相調(diào)用的詳細內(nèi)容，更多關于Python grpc服務的資料請關注腳本之家其它相關文章！
相關文章
Python3.4學習筆記之常用操作符,條件分支和循環(huán)用法示例
這篇文章主要介紹了Python3.4常用操作符,條件分支和循環(huán)用法,結合實例形式較為詳細的分析了Python3.4常見的數(shù)學運算、邏輯運算操作符,條件分支語句,循環(huán)語句等功能與基本用法,需要的朋友可以參考下2019-03-03
解決Python import .pyd 可能遇到路徑的問題
這篇文章主要介紹了解決Python import .pyd 可能遇到路徑的問題，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2021-03-03
這篇文章主要介紹了python Graham求凸包問題并畫圖操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-06-06
最近完成的一個項目用到了SVM，之前也一直有聽說支持向量機，知道它是機器學習中一種非常厲害的算法。利用將近一個星期的時間學習了一下支持向量機，把原理推了一遍，感覺支持向量機確實挺厲害的，這篇文章帶你了解它2022-01-01
python pycurl驗證basic和digest認證的方法
這篇文章主要介紹了python pycurl驗證basic和digest認證的方法，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-05-05
Python實現(xiàn)Excel做表自動化的最全方法合集
Microsoft?Excel?是一款強大的辦公工具,廣泛用于數(shù)據(jù)分析、報告制作、預算管理等各種任務,本文將深入探討如何使用?Python?進行?Excel?表格的自動化,需要的可以參考下2024-02-02
最近在工作中遇到一個坑，發(fā)現(xiàn)這個坑居然存在于return和finally，所以覺著有必要總結分享一下，下面這篇文章主要介紹了關于Python中return和finally共同挖的坑，需要的朋友可以參考借鑒，下面來一起看看吧。2017-08-08
在我們生活中的一些場合經(jīng)常會有一些不該出現(xiàn)的敏感詞,我們通常會使用*去屏蔽它,一些罵人的敏感詞和一些政治敏感詞都不應該出現(xiàn)在一些公共場合中,這個時候我們就需要一定的手段去屏蔽這些敏感詞,下面我來介紹一些簡單版本的Python敏感詞屏蔽的方法,需要的朋友可以參考下2025-04-04
使用Pytest.main()運行時參數(shù)不生效問題解決
本文主要介紹了使用Pytest.main()運行時參數(shù)不生效問題解決，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-02-02
Python cookbook（數(shù)據(jù)結構與算法）將序列分解為單獨變量的方法
這篇文章主要介紹了Python cookbook（數(shù)據(jù)結構與算法）將序列分解為單獨變量的方法,結合實例形式分析了Python序列賦值實現(xiàn)的分解成單獨變量功能相關操作技巧,需要的朋友可以參考下2018-02-02