什么是PB協議:Protocol Buffer是Google提供的一種數據序列化協議:
-
穩定可維護的Node socket.io服務(socket篇)
-
PB的動態編譯的特性和嵌套message,數據結構簡單化(數據篇)
-
前端對於二級制的的處理,例如ArrayBuffer,序列化與反序列化,MD5加密(加密篇)
1.其實不用管他是什么協議,我們只需要明白,作為測試他能否實現業務需要的功能,然后能否抗住壓力 (基本功能+性能壓力)
2.proto文件 需要轉化為python的協議文件 ,具體方式 是 pip install protobuf
3.安裝了這個包 可以pip show protobuf 看一下
4.生成命令:protoc --proto_path=PROTO_PATH --python_out=OUT_DIR xxx.proto
5.生成python的協議文件之后,我們需要創建好請求參數 json格式就行
data = {
"data": [
{
"f1": 1.0,
"f2": 1.0,
"f3": 1,
"f4": 1,
"f5": 1,
"f6": 1,
"f7": "aa",
"f8": "12222222",
"f9": 12222222,
"f18": "aa",
"f19": "12222222",
"f12": "aa",
"f13": "aa",
"f10": 1,
"f11": "com.android.browser",
"f16": "aa",
"f17": "aa",
"f14": "aa",
"f15": "aa"
}
]
}
例如;如上的json 通過json to pb來轉化我們需要的pb格式
from google.protobuf.json_format import ParseDict
da = ParseDict(js_dict=data, message=LogSyncMessage())
data = da.SerializeToString() # 進行序列化
然后requests 發起請求就ok了
6.對於一些大數據量的傳輸,可能需要將序列化之后的數據 進行gzip進行壓縮傳輸請求
列出壓縮的函數:
def Testzip(c_data):
buf = StringIO()
try:
with gzip.GzipFile(mode='wb', fileobj=buf) as f:
f.write(c_data)
return buf.getvalue()
except Exception as e:
print ("compress wrong" + e)
finally:
f.close()
發起請求 將json進行轉化為pb格式 然后進行序列化 然后利用公共函數 進行gzip進行壓縮 然后發起請求
data = da.SerializeToString() # 序列化
data = self.Testzip(data)
r = requests.post(url=url, data=data, headers=self.header)
完成