接到一个应用测试，应用实现主要使用websocket保持长链接，使用protobuf格式交换数据，用途为发送消息，需要我们测试评估性能，初步评估需要测试长链接数、峰值消息数以及长期运行稳定性
整体需求不难，主要困难是第一次接触websocket和protobuf，同时也没有现成的脚本可以使用，jmeter原生也不支持websocket协议，需要安装插件支持，protobuf格式也需要重新编译大包使用
万里长征第一步，先开发测试脚本，由于目前一直在使用python开发，评估项目的性能要求，不需要严格性能指标，所以直接使用python作为开发语言编写脚本，主要实现脚本如下
入口主文件，mulitprocess.py，多进程发起建立链接
1 # -*- coding: utf-8 -*- 2 # @Author : monleylu 3 # @Time : 2018/7/5 12:22 PM 4 5 from multiprocessing import Process,Pool 6 import os 7 from data import user 8 from protobufcon import ProtoService 9 import tnlog,logging 10 11 def consumer(host,port,username): 12 logging.error('启动子进程： %s...' % (os.getpid())) 13 kwargs = {"host": host, "port": port, "username": username } 14 pro=ProtoService(**kwargs) 15 pro.startWebsocket() 16 17 18 def consumer2(*args,**kwargs): 19 logging.error('启动子进程： %s...' % (os.getpid())) 20 pro = ProtoService(**kwargs) 21 pro.startWebsocket() 22 23 24 25 if __name__ =="__main__": 26 logging.error('启动主进程 %s.' % os.getpid()) 27 28 p = Pool(450) 29 for i in range(450): 30 #第一种传参数方法 31 # p.apply_async(consumer, args=("11.31.60.10",8880,user.pop(),)) 32 33 #第二种传参数方法 34 kw={"host": "11.31.60.10", "port": 8880, "username": user.pop()} 35 p.apply_async(consumer, args=(),kwds=kw) 36 37 logging.error('Waiting for all subprocesses done...') 38 39 p.close() 40 p.join() 41 logging.error('All subprocesses done.')
实现代码，protobuffun.py，主要负责发送、接收消息
1 # -*- coding: utf-8 -*- 2 # @Author : monleylu 3 # @Time : 2018/7/5 12:15 PM 4 5 import protoinfo.msg_pb2 as msgpb2 6 import websocket,json,datetime 7 import json 8 try: 9 import thread 10 except ImportError: 11 import _thread as thread 12 import tnlog,logging 13 from data import user 14 class ProtoService: 15 16 def __init__(self,*,host="10.30.66.11",port=8886,username="monleylu"): 17 self.host=host 18 self.port=port 19 self.username=username 20 21 def on_message(self,ws, message): 22 msg2 = msgpb2.Msg() 23 msg2.ParseFromString(message) 24 logging.error("用户:{} 收到消息，类型:{},内容:{}".format(self.username,msg2.method,msg2.data)) 25 if msg2.method == "login": 26 b=msg2.data 27 c=json.loads(b) 28 self.sendprotobuf(ws,"serve_get",{'uc_id':c["uc_id"],'boxReadFlag':'0','boxArriveFlag':'0','boxCollectFlag':'0'}) 29 30 def on_error(self,ws, error): 31 logging.error("发生异常:{}".format(error)) 32 33 def on_close(self,ws): 34 ws.close() 35 logging.error("### closed ###") 36 37 38 # def wrapp(username): 39 # def w(*args,**kwargs): 40 41 42 def on_open(self,ws): 43 self.sendprotobuf(ws, "connect", {'uc_name': self.username}) 44 45 def sendprotobuf(self,ws,method,data): 46 msg = msgpb2.Msg() 47 msg.method = method 48 msg.data = json.dumps(data) 49 ws.send(msg.SerializeToString()) 50 logging.error("发送消息，类型:{},内容:{}".format(msg.method, msg.data)) 51 52 53 def startWebsocket(self): 54 # websocket.enableTrace(True) 55 logging.error("连接服务器:{},端口：{},用户名:{}".format(self.host,self.port,self.username)) 56 ws = websocket.WebSocketApp("ws://{}:{}".format(self.host,self.port), 57 on_message=self.on_message, 58 on_error=self.on_error, 59 on_close=self.on_close) 60 ws.on_open = self.on_open 61 62 # ws.run_forever(ping_interval=5) 63 logging.error(datetime.datetime.now()) 64 ws.run_forever(ping_interval=5) 65 # ws.run_forever() 66 logging.error(datetime.datetime.now()) 67 68 69 #10.30.66.11 70 71 if __name__ == "__main__": 72 kwargs = {"host": "10.30.66.11", "port": 8886, "username": "liangyuanfang", } 73 p=ProtoService(**kwargs) 74 p.startWebsocket()
查询运行过程中，会把日志写到对应的文件中，通过统计文件内容分析测试结果
这里可能各位也注意到写日志，代码使用了error级别，因为之前在调试时，info或者debug级别也会打印出一些引用组建的调试代码，通过这种方式规避下