Python传输解析PB消息Client段程序

怼烎@ 2022-07-12 07:44 122阅读 0赞

参考链接

*  [http://blog.csdn.net/huithe/article/details/5223785][http_blog.csdn.net_huithe_article_details_5223785]

程序中使用python直接通过protobuf协议访问服务器，当服务器返回的数据比较大的时候，client端在接收数据之前，需要sleep一段时间，否则接收到的数据会不完整，然后解析出错。

pb文件内容如下：

package bse.bsp.triggerserver;
    
    message TriggerRequest {
        required uint32 client_id    = 1;    /*请求id*/
        required uint64 idea_id      = 2;    /*idea_id*/
        required bytes title         = 3;    /*原始title字面*/
    };
    
    message TriggerInfo {
        required uint64 query_sign   = 1;
        required bytes relv_score    = 2;
        required bytes query         = 3;
        required uint64 wmatch       = 4;     /*一个query被多种方式挖掘到*/
        required uint64 fea_sign     = 5;     /*所有特征拼接在一起的签民*/
    };
    
    message TriggerResponse {
        required uint64 idea_id      = 1;
        required uint32 update_time  = 2;    /*更新时间*/
        repeated TriggerInfo trigger_info_list = 3; /*query信息*/
    };

生成pb的python依赖包

protoc -I=. --python_out=../interface ./trigger_server.proto

client端代码如下：

# !/usr/bin/env python
        # -*- coding:gbk -*-
    
        import struct
        import socket
        import time
    
        import sys
        sys.path.append('./interface/')
        import trigger_server_pb2
    
        class TriggerClient(object):
    
            def __init__(self, ip, port):
                self.ip = ip
                self.port = port
                self.address = (ip, port)
                self.id = 0
                self.version = 0
                self.log_id = 0
                self.provider = "trigger_server"
                self.magic_num = int('949370fb', 16)
                self.reserverd = 0
                self.buf_size = 102400
    
            def talk(self, title):
                request = trigger_server_pb2.TriggerRequest()
                request.client_id = 1
                request.idea_id = 123456
                request.title = title
    
                try:
                    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                    sock.connect(self.address)
    
                    msg = request.SerializeToString()
                    body_len = struct.unpack('<I', struct.pack('>I', len(msg)))[0]
                    nshead = struct.pack('!HHI16sIII',
                                         self.id,
                                         self.version,
                                         self.log_id,
                                         self.provider,
                                         self.magic_num,
                                         self.reserverd,
                                         body_len)
                    packet = nshead + msg
                    sock.send(packet)
    
                    time.sleep(0.5)
                    data = sock.recv(self.buf_size)
                    response = trigger_server_pb2.TriggerResponse()
                    body = data[len(nshead):]
                    response.ParseFromString(body)
                    for i in range(len(response.trigger_info_list)):
                        query = response.trigger_info_list[i].query
                        print query
                except Exception, e:
                    print e
                    print "exception occur"
    
        def main():
            relv_client = TriggerClient('10.95.22.53', 8757)
            input = sys.argv[1]
            output = sys.argv[2]
    
            with open(input, 'r') as fin, open(output, 'w') as fout:
                for line in fin:
                    parts = line.rstrip().split('\\t')
                    if len(parts) != 1:
                        continue
                    title = parts[0]
                    relv_client.talk(title)
    
        if __name__ == "__main__":
            main()

[http_blog.csdn.net_huithe_article_details_5223785]: http://blog.csdn.net/huithe/article/details/5223785