初衷：grpc可以很好的作为跨语言，跨机器的服务模式。但是对于大量数据的传输，进程之间还是共享内存要快一些，所以这里将grpc与共享内存混合使用。

功能实现：

        客户端：从摄像头不断的读取图片，并传输给服务端，并呈现从服务端返回的处理结果。

        服务端：读取客户端的请求，并从中读取共享内存中的图片，在图片上写入”ok“，并将结果传回给客户端。

创建文件：image_processing.proto

syntax = "proto3";package image_processing;import "google/protobuf/empty.proto";// 定义一个请求消息，包含共享内存的名称
message ImageRequest {string input_shm_name = 1;string output_shm_name = 2;
}// 定义一个响应消息，包含处理后的图像的共享内存地址
message ImageResponse {string output_shm_name = 1;
}// 定义清理共享内存的请求消息
message CleanupRequest {string output_shm_name = 1;
}// 定义服务接口
service ImageProcessingService {rpc ProcessImage (ImageRequest) returns (ImageResponse);rpc CleanupSharedMemory (CleanupRequest) returns (google.protobuf.Empty);
}

安装python库：

pip install grpcio grpcio-tools opencv-python-headless

使用grpcio-tools生成Python代码：

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. image_processing.proto

客户端

import grpc
import cv2
import numpy as np
from multiprocessing import shared_memoryimport image_processing_pb2
import image_processing_pb2_grpc
def main():# 捕获图像#img = capture_image()vc=cv2.VideoCapture(0)with grpc.insecure_channel('localhost:50051') as channel:stub = image_processing_pb2_grpc.ImageProcessingServiceStub(channel)while 1:res,img=vc.read()shm = shared_memory.SharedMemory(create=True, size=img.nbytes)shm_array = np.ndarray(img.shape, dtype=img.dtype, buffer=shm.buf)shm_array[:] = img[:]response = stub.ProcessImage(image_processing_pb2.ImageRequest(input_shm_name=shm.name, output_shm_name=""))# 读取处理后的图像output_shm = shared_memory.SharedMemory(name=response.output_shm_name)output_array = np.ndarray(img.shape, dtype=img.dtype, buffer=output_shm.buf)# 显示处理后的图像cv2.imshow("Processed Image", output_array)cv2.waitKey(1)cv2.destroyAllWindows()# 清理共享内存shm.close()shm.unlink()output_shm.close()output_shm.unlink()# 通知服务端清理共享内存with grpc.insecure_channel('localhost:50051') as channel:stub = image_processing_pb2_grpc.ImageProcessingServiceStub(channel)stub.CleanupSharedMemory(image_processing_pb2.CleanupRequest(output_shm_name=response.output_shm_name))if __name__ == '__main__':main()

服务端：

import grpc
from concurrent import futures
import cv2
import numpy as np
from multiprocessing import shared_memory
import time
import threadingimport image_processing_pb2
import image_processing_pb2_grpc# 全局字典存储共享内存对象引用
shared_memory_store = {}
lock = threading.Lock()class ImageProcessingServiceServicer(image_processing_pb2_grpc.ImageProcessingServiceServicer):def ProcessImage(self, request, context):# 读取共享内存中的图像shm = shared_memory.SharedMemory(name=request.input_shm_name)img_array = np.ndarray((480, 640, 3), dtype=np.uint8, buffer=shm.buf)img = img_array.copy()  # 复制图像以确保共享内存可以被释放# 在图像上写“ok”cv2.putText(img, "ok", (50, 50), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 0, 0), 2, cv2.LINE_AA)# 创建新的共享内存并将处理后的图像写入output_shm = shared_memory.SharedMemory(create=True, size=img.nbytes)output_array = np.ndarray(img.shape, dtype=img.dtype, buffer=output_shm.buf)output_array[:] = img[:]# 将共享内存对象存储在全局字典中with lock:shared_memory_store[output_shm.name] = output_shm# 启动一个线程在一段时间后清理共享内存threading.Timer(60, self.cleanup_shared_memory, args=(output_shm.name,)).start()# 返回处理后的图像的共享内存名称response = image_processing_pb2.ImageResponse(output_shm_name=output_shm.name)return responsedef cleanup_shared_memory(self, shm_name):with lock:if shm_name in shared_memory_store:shm = shared_memory_store.pop(shm_name)shm.close()shm.unlink()def serve():server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))image_processing_pb2_grpc.add_ImageProcessingServiceServicer_to_server(ImageProcessingServiceServicer(), server)server.add_insecure_port('[::]:50051')server.start()print("Server started, listening on port 50051.")try:while True:time.sleep(86400)  # 运行一整天except KeyboardInterrupt:server.stop(0)if __name__ == '__main__':serve()