[日常] gRPC 踩坑记录（Python）

红太狼 2021-10-31 15:52 484阅读 0赞

> gRPC is a modern open source high performance RPC framework that can run in any environment. It can efficiently connect services in and across data centers with pluggable support for load balancing, tracing, health checking and authentication. It is also applicable in last mile of distributed computing to connect devices, mobile applications and browsers to backend services.

### 1. gRPC 是什么 ###

> 官方文档：[链接][Link 1]，中文文档：[链接][Link 2]，GitHub 网址：[链接][Link 3]。

想了解 gRPC 是什么，就得先了解什么是 RPC。RPC 全称为 Remote Procedure Call，即远程调用框架（从字面意思上看就是“远程调用服务端所提供的方法/函数”），该协议允许运行于一台计算机的程序调用另一台计算机的子程序，而程序员无需额外地为这个交互作用编程。

简单来说就是，通过利用 RPC，我们可以轻松调用服务端所提供的函数，而无需了解其内部定义。（关于RPC、REST API、Socket 的区别，可以查看这篇博文：[RPC是什么？RPC与REST、Socket的区别？php中流行的rpc框架有哪些？][RPC_RPC_REST_Socket_php_rpc]）

那么 gRPC 是什么呢？gRPC是一个高性能、通用的开源RPC框架，其由 Google 主要面向移动应用开发并基于HTTP/2协议标准而设计，基于 ProtoBuf(Protocol Buffers) 序列化协议开发，且支持众多开发语言。 其最基本的开发步骤是定义 proto 文件， 定义请求 Request 和 响应 Response 的格式，然后定义一个服务 Service， Service 可以包含多个方法。

![20190816110856357.png][]

此外，gRPC 支持流式开发，即客户端流式发送、服务器流式返回以及客户端／服务器同时流式处理, 也就是单向流和双向流。(关于流式开发，可以阅读这篇博文：[gRPC的那些事 - streaming][gRPC_ - streaming])

![20190816111607499.png][]

### 2. 安装 gRPC ###

# gRPC 的安装：
    $ pip install grpcio
    
    # 安装 ProtoBuf 相关的 python 依赖库：
    $ pip install protobuf
    
    # 安装 python grpc 的 ProtoBuf 编译工具：
    $ pip install grpcio-tools

### 3. gRPC 简单例子 ###

*  编写 proto 文件并进行编译

syntax = "proto3";
    
    message EmotionClsRequest {
        bytes image = 1;
    
        message Shape {
            int32 height = 1;
            int32 width = 2;
            int32 channel = 3;
        }
    
        Shape shape = 2;
    }
    
    message EmotionClsResponse {
        message Box {
            int32 left = 1;
            int32 top = 2;
            int32 right = 3;
            int32 bottom = 4;
        }
    
        message Emotion {
            float Surprised = 1;
            float Fear = 2;
            float Disgust = 3;
            float Happy = 4;
            float Sad = 5;
            float Angry = 6;
            float Neutral = 7;
    
        }
    
        message Face {
            Box box = 1;
            float score = 2;
            Emotion emotion = 3;
            int32 errorCode = 4;
        }
    
        repeated Face faces = 1;
    }
    
    service EmotionClassifier {
        rpc Classification (EmotionClsRequest) returns (EmotionClsResponse) {}
    }

可以看到，在这个proto 文件中，我们定义了消息 EmotionClsRequst，消息 EmotionClsResponse，服务 EmotionClassifier 以及其中的 RPC 方法 Classification 的信息格式（相关语法可查看后续的 Protocol Buffers 介绍）。此外，我们在此所定义的服务中 Classification 为简单的 RPC 方法，若想定义流式 RPC 方法，可这样编写：

service EmotionClassifier{
    	rpc  Classification1(EmotionClsRequest) returns (stream EmotionClsResponse){} // 应答流式 RPC ， 客户端发送请求到服务器，拿到一个流去读取返回的消息序列。 客户端读取返回的流，直到里面没有任何消息。
    	rpc  Classification2(stream EmotionClsRequest) returns (stream EmotionClsResponse){} // 请求流式 RPC ， 客户端写入一个消息序列并将其发送到服务器，同样也是使用流。一旦客户端完成写入消息，它等待服务器完成读取返回它的响应。
    	rpc  Classification3(stream EmotionClsRequest) returns (stream EmotionClsResponse){} // 双向流式 RPC 是双方使用读写流去发送一个消息序列。
    }

接着，对 proto 文件进行编译：

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

该命令将在当前目录下生成 emotionCls\_pb2.py 和 emotionCls\_pb2\_grpc.py。其中，–python\_out 指定所输出的 \*\_pb2.py 文件路径，–grpc\_python\_out 指定所输出的 \*\_pb2\_grpc.py 文件路径，-I 是指定.proto文件所在路径。

*  编写服务端（server）代码

from concurrent import futures
    
    import os
    import sys
    import time
    import copy
    import logging
    import argparse
    import numpy as np
    
    import grpc
    import emotionCls_pb2
    import emotionCls_pb2_grpc
    
    _ONE_DAY_IN_SECONDS = 60 * 60 * 24
    
    sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
    # sys.path.append(os.path.join(os.path.dirname(__file__), ".."))
    
    from face_detection.face_detector import FaceDetector
    from face_detection.config import cfg as face_detection_cfg
    
    from face_align.face_aligner import FaceAligner
    from face_align.config import cfg as face_alignment_cfg
    
    from face_expression.face_expression_cls import FaceExpressionCls
    from face_expression.config import cfg as face_expression_cfg
    
    parser = argparse.ArgumentParser()
    parser.add_argument("--detection_config_file",
                        default="../../configs/face_detection_gRPC.yaml",
                        help="path to yaml config file", type=str)
    parser.add_argument("--alignment_config_file",
                        default="../../configs/face_alignment_gRPC.yaml",
                        help="path to yaml config file", type=str)
    parser.add_argument("--expression_config_file",
                        default="../../configs/face_expression_gRPC.yaml",
                        help="path to yaml config file", type=str)
    parser.add_argument("opts", help="modify config options from commandline",
                        default=None, nargs=argparse.REMAINDER)
    args = parser.parse_args()
    
    if args.detection_config_file != "":
        face_detection_cfg.merge_from_file(args.detection_config_file)
    face_detection_cfg.merge_from_list(args.opts)
    face_detection_cfg.freeze()
    
    if args.alignment_config_file != "":
        face_alignment_cfg.merge_from_file(args.alignment_config_file)
    face_alignment_cfg.merge_from_list(args.opts)
    face_alignment_cfg.freeze()
    
    if args.expression_config_file != "":
        face_expression_cfg.merge_from_file(args.expression_config_file)
    face_expression_cfg.merge_from_list(args.opts)
    face_expression_cfg.freeze()
    
    face_detector = FaceDetector(face_detection_cfg)
    face_aligner = FaceAligner(face_alignment_cfg)
    face_expression_cls = FaceExpressionCls(face_expression_cfg)
    
    # 继承 emotionCls_pb2_grpc.py 中的 EmotionClassifierServicer 类
    # 完成在 emotionCls.proto 文件中所声明的服务的定义
    class Classifier(emotionCls_pb2_grpc.EmotionClassifierServicer): 
    
        def Classification(self, request, context): # 业务逻辑
            image = np.array(list(request.image), dtype=np.uint8)
            image = np.reshape(image, (request.shape.height, request.shape.width, request.shape.channel))
    
            faces = []
    
            # Get bounding box.
            bboxes = face_detector.detect_faces(image.copy())
            if bboxes is None or len(bboxes)==0:
                response = emotionCls_pb2.EmotionClsResponse(faces=faces)
                return response
    
            for index in range(bboxes.shape[0]):
                # Get landmark.
                bbox = [np.int32(bboxes[index,0]), np.int32(bboxes[index,1]), np.int32(bboxes[index,2]), np.int32(bboxes[index,3])]
                landmark = face_aligner.detect_landmark(image.copy(), copy.deepcopy(bbox))
                if landmark==np.array([]) or landmark.size==0:
                    box = emotionCls_pb2.EmotionClsResponse.Box(left=np.int32(bboxes[index,0]),top=np.int32(bboxes[index,1]),right=np.int32(bboxes[index,2]),bottom=np.int32(bboxes[index,3]))
                    emotion = emotionCls_pb2.EmotionClsResponse.Emotion(Surprised=0,Fear=0,Disgust=0,Happy=0,Sad=0,Angry=0,Neutral=0)
                    face = emotionCls_pb2.EmotionClsResponse.Face(box=box,score=bboxes[index,4],emotion=emotion,errorCode=201)
                    
                    faces.append(face)
                    continue
    
                # Get expression.
                pred = face_expression_cls.predict_expression(*face_aligner.align_face(image.copy(),copy.deepcopy(landmark)))
    
                box = emotionCls_pb2.EmotionClsResponse.Box(left=np.int32(bboxes[index,0]),top=np.int32(bboxes[index,1]),right=np.int32(bboxes[index,2]),bottom=np.int32(bboxes[index,3]))
                emotion = emotionCls_pb2.EmotionClsResponse.Emotion(Surprised=pred['Surprised'],Fear=pred['Fear'],Disgust=pred['Disgust'],Happy=pred['Happy'],Sad=pred['Sad'],Angry=pred['Angry'],Neutral=pred['Neutral'])
                face = emotionCls_pb2.EmotionClsResponse.Face(box=box,score=bboxes[index,4],emotion=emotion,errorCode=202)
    
                faces.append(face)
    
            response = emotionCls_pb2.EmotionClsResponse(faces=faces)
            return response
    
    
    def serve():
    
        print('Server Run...')
    
        server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) # 定义服务器，并设置最大连接数
        emotionCls_pb2_grpc.add_EmotionClassifierServicer_to_server(Classifier(), server) # 添加服务
        server.add_insecure_port('[::]:50050') # 设置服务器监听端口号
        server.start()
        try:
            while True:
                time.sleep(_ONE_DAY_IN_SECONDS)
        except KeyboardInterrupt:
            server.stop(0)
    
    
    if __name__ == '__main__':
        logger = logging.getLogger()
        logger.setLevel(logging.INFO)
        handler = logging.StreamHandler(stream=sys.stdout)
        logger.addHandler(handler)
        serve()

*  编写客户端（client）代码

import os
    import cv2
    import grpc
    import logging
    import numpy as np
    
    import emotionCls_pb2
    import emotionCls_pb2_grpc
    
    _HOST = '192.168.3.3'
    _PORT = '50050'
    
    Label = ['Surprised', 'Fear', 'Disgust', 'Happy', 'Sad', 'Angry', 'Neutral']
    
    def run():
        path_list = ['test_image/01.jpeg', 'test_image/02.jpeg', 'test_image/03.jpeg']
    
        with grpc.insecure_channel(_HOST + ':' + _PORT) as channel:
            stub = emotionCls_pb2_grpc.EmotionClassifierStub(channel) # 创建与服务端的连接
    
            for path in path_list:
                img = cv2.imread(path)
                height, width, channel = img.shape
                shape = emotionCls_pb2.EmotionClsRequest.Shape(height=height, width=width, channel=channel)
                request = emotionCls_pb2.EmotionClsRequest(image=bytes(img), shape=shape) # 依据 emotionCls.proto 所定义的格式，创建想要的 Request
                response = stub.Classification(request) # 获取由服务端所返回的 Response
    
                for face in response.faces:
                    cv2.rectangle(img, (face.box.left, face.box.top),
                                (face.box.right, face.box.bottom), (0, 255, 0), 2)
                    preds = np.array([face.emotion.Surprised, face.emotion.Fear, face.emotion.Disgust, face.emotion.Happy, face.emotion.Sad, face.emotion.Angry, face.emotion.Neutral])
                    index = np.argmax(preds)
    
                    cv2.putText(img, Label[index]+':'+str(preds[index]), (face.box.left, face.box.top - 5),
                                    cv2.FONT_HERSHEY_SIMPLEX, 1, (0,0,255), 1, cv2.LINE_AA)
                    print('Error Code: %d' % (face.errorCode))
    
                cv2.imshow('test', img)
                if cv2.waitKey(0) == 27:
                    break
    
    
    if __name__ == '__main__':
        logging.basicConfig()
        run()

### 4. 关于 Protocol Buffers ###

Protocol Buffers 是一种轻便高效的结构化数据存储格式（由 Google 提供的一种数据序列化协议），可以用于结构化数据序列化，很适合做数据存储或 RPC 数据交换格式。它可用于通讯协议、数据存储等领域的语言无关、平台无关、可扩展的序列化结构数据格式。

如果想了解 Protocol Buffers 具体语法，建议大家阅读以下博文：

*  [\[译\]Protobuf 语法指南][Protobuf]
 *  [Protocol Buffers 3.0 技术手册][Protocol Buffers 3.0]
 *  [Protocol Buffers Developer Guide][]

**参考资料：**

*  [谁能用通俗的语言解释一下什么是 RPC 框架？][RPC]
 *  [RPC 是什么][RPC 1]
 *  [gRPC是什么][gRPC]

[Link 1]: https://grpc.io/
[Link 2]: https://doc.oschina.net/grpc?t=56831
[Link 3]: https://github.com/grpc/grpc
[RPC_RPC_REST_Socket_php_rpc]: https://blog.csdn.net/cs958903980/article/details/78674087
[20190816110856357.png]: /images/20211031/1ada979e15224eeea51b3d5737770a0d.png
[gRPC_ - streaming]: https://colobu.com/2017/04/06/dive-into-gRPC-streaming/
[20190816111607499.png]: /images/20211031/72e9d04384d345738d0e1a133fa1d956.png
[Protobuf]: https://colobu.com/2015/01/07/Protobuf-language-guide/#%E6%B7%BB%E5%8A%A0%E6%B3%A8%E9%87%8A
[Protocol Buffers 3.0]: https://blog.csdn.net/shensky711/article/details/69696392
[Protocol Buffers Developer Guide]: https://developers.google.com/protocol-buffers/docs/pythontutorial
[RPC]: https://www.zhihu.com/question/25536695
[RPC 1]: https://toutiao.io/posts/boebs3/preview
[gRPC]: https://www.kancloud.cn/runningday/grpc/265663