Skip to content

N
new-socketemulator
Commit f0bb02f9 authored by liyuanhong's avatar liyuanhong
Browse files
Options

M300 压测程序完成

parent c4fd6987
Hide whitespace changes
Inline Side-by-side
Showing with 306 additions and 60 deletions
lib/multiThread/SendMultMsgThread_m300.py
View file @ f0bb02f9
#coding:utf-8
import binascii
import json
import random
import socket
import threading
import time
......@@ -15,12 +16,21 @@ class SendMultMsgThread():
self.host = host
self.port = port
self.msg = msg
self.timeOut = 1 #socket超时时间
self.timeOut = 30 #socket超时时间
self.BUF_SIZE = 1024 #接收消息缓存
self.threadCount = 10 #并发线程数
self.threadCount = 10000 #并发线程数
self.totalTime = 0 #所有线程的运行总和
self.threadArr = {} #保存每个线程的信息
self.failThreadCount = 0 #失败线程数
self.durThreads = [] #持续发送线程数组，当数组为空，表示所有线程已经结束
dt = 1 * 1 * 60
self.durTime = dt #线程持续时间
self.connectTimeoutNum = 0 #连接超时线程数
self.sendTimeoutNum = 0 #发送超时线程数
self.reviceTimeoutNum = 0 #接收超时线程数
self.sucessNum = 0 #成功线程数
self.messageCon = [] #用来统计每个线程所发的消息数
self.messageCons = 0 # 用来统计每个线程所发的消息数
pass
############################################
......@@ -79,20 +89,87 @@ class SendMultMsgThread():
self.totalTime = self.totalTime + timeExpend
client.close()
############################################
# 持续发送消息
# dur: 持续发送时间，如果为0一直发送， 设置了事件为持续发送多少秒，默认为一分钟
############################################
def sendMsgContinuous(self,carId,threadName="thread0"):
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) # 在客户端开启心跳
client.settimeout(self.timeOut)
startTime = int(time.time())
endTime = int(time.time())
msgCon = 0 #统计线程发的消息数量
self.durThreads.append(threadName)
try:
client.connect((self.host, self.port))
except BaseException as e:
client.close()
self.durThreads.remove(threadName)
self.threadArr[threadName]["status"] = 1
self.failThreadCount = self.failThreadCount + 1
endTime = int(time.time())
timeExpend = endTime - startTime
self.threadArr[threadName]["timeExp"] = timeExpend
self.connectTimeoutNum = self.connectTimeoutNum + 1
self.messageCon.append(msgCon)
self.threadArr[threadName]["msgCon"] = msgCon
print(threadName + ":" + "连接超时，socket断开")
return
while (endTime - startTime) < self.durTime:
msg = self.getRandomMsg_M300(carId)
try:
client.send(binascii.a2b_hex(msg))
msgCon = msgCon + 1
self.messageCons = self.messageCons + 1
except BaseException as e:
client.close()
self.durThreads.remove(threadName)
self.threadArr[threadName]["status"] = 1
self.failThreadCount = self.failThreadCount + 1
endTime = int(time.time())
timeExpend = endTime - startTime
self.threadArr[threadName]["timeExp"] = timeExpend
self.sendTimeoutNum = self.sendTimeoutNum + 1
self.messageCon.append(msgCon)
self.threadArr[threadName]["msgCon"] = msgCon
print(threadName + ":" + "发送超时，socket断开")
return
try:
data = client.recv(self.BUF_SIZE)
except BaseException as e:
client.close()
self.durThreads.remove(threadName)
self.threadArr[threadName]["status"] = 1
self.failThreadCount = self.failThreadCount + 1
self.reviceTimeoutNum = self.connectTimeoutNum + 1
self.messageCon.append(msgCon)
self.threadArr[threadName]["msgCon"] = msgCon
print(threadName + ":" + 'socket 接收消息超时！')
endTime = int(time.time())
timeExpend = endTime - startTime
self.threadArr[threadName]["timeExp"] = timeExpend
return
endTime = int(time.time())
sleepTime = random.randint(1, 5)
time.sleep(sleepTime)
endTime = int(time.time())
timeExpend = endTime - startTime
self.threadArr[threadName]["timeExp"] = timeExpend
self.messageCon.append(msgCon)
self.threadArr[threadName]["msgCon"] = msgCon
client.close()
self.sucessNum = self.sucessNum + 1
self.durThreads.remove(threadName)
############################################
# 启动并发线程
############################################
def startThread(self):
timeStart = int(time.time() * 1000)
for i in range(0,self.threadCount):
threadName = "thread" + str(i)
# theThread = threading.Thread(target=self.sendMsg, args=("7e0002000001314620111800065b7e",threadName,)) # 数据写死，心跳
# theThread = threading.Thread(target=self.sendMsg,args=("7e0200010908597573774756702ed057ac0016fb16010f4f5208e98691021700400024951202144709EA6600010400000000000204001e7c1f0003050A0001f400000405020001d4c000050400057d0240000604000119400007040007530000100c0004006403f203f203f203f2001114ffffffffffffffffffff00200000000000000000001202002400130106001D0101310160e07e", threadName,)) # 数据写死
# theThread = threading.Thread(target=self.sendMsg, \
# args=(TerminalRegister_msg().generateMsg_random(), threadName,)) #终端注册
theThread = threading.Thread(target=self.sendMsg, \
args=(Location_msg().generateMsg_random(), threadName,)) #地理位置
threadName = "thread-" + str(i)
theThread = threading.Thread(target=self.sendMsg, args=("7e0002000001314620111800065b7e",threadName,)) # 数据写死，心跳
threadInfo = {}
threadInfo["name"] = threadName
threadInfo["status"] = 0
......@@ -100,42 +177,188 @@ class SendMultMsgThread():
theThread.start()
timeEnd = int(time.time() * 1000)
timeExpend = timeEnd - timeStart
time.sleep(20)
time.sleep(3)
print("耗时：" + str(timeExpend) + " 毫秒")
print("并发数据每秒发送：" + str(int(self.threadCount / (timeExpend / 1000))))
print("平均响应时间：" + str(self.totalTime / self.threadCount) + "毫秒")
print("发送总数：" + str(self.threadCount))
print("响应失败数：" + str(self.failThreadCount))
self.writeToFile("../../data/threadDetails.json",self.threadArr)
# print(json.dumps( self.threadArr))
############################################
# 启动持续并发线程
############################################
def startThreadContinuous(self):
timeStart = int(time.time() * 1000)
for i in range(0,self.threadCount):
threadName = "thread-" + str(i)
print(threadName)
carid = 201912000000 + i
theThread = threading.Thread(target=self.sendMsgContinuous, args=(carid,threadName,)) # 数据写死，心跳
threadInfo = {}
threadInfo["name"] = threadName
threadInfo["status"] = 0
self.threadArr[threadName] = threadInfo
theThread.start()
timeEnd = int(time.time() * 1000)
timeExpend = timeEnd - timeStart
print("耗时：" + str(timeExpend) + " 毫秒产生了" + str(self.threadCount) + "线程")
time.sleep(0.5) #防止启动的时候溜掉某些启动比较慢的线程
tmp = 1
while len(self.durThreads) != 0:
print("剩余线程数_" + str(tmp) + "：" + str(len(self.durThreads)))
tmp = tmp + 1
timeArray = time.localtime(timeStart / 1000)
testStart = time.strftime("%Y-%m-%d %H:%M:%S", timeArray)
timeCur = int(time.time() * 1000)
timeArray = time.localtime(timeCur / 1000)
testCur = time.strftime("%Y-%m-%d %H:%M:%S", timeArray)
info_0 = "-------------------------- 统计信息（pre） --------------------------"
info_1 = "耗时：" + str(timeExpend) + " 毫秒产生了" + str(self.threadCount) + "线程"
info_2 = "开始测试时间：" + testStart
info_4 = "设置socket超时时间：" + str(self.timeOut)
info_5 = "设置线程持续时间：" + str(self.durTime)
info_6 = "剩余线程数：" + str(len(self.durThreads))
info_8 = "连接失败：" + str(self.connectTimeoutNum)
info_9 = "发送失败：" + str(self.sendTimeoutNum)
info_10 = "接收失败：" + str(self.reviceTimeoutNum)
info_11 = "当前写入时间：" + testCur
info_12 = "当前发送消息总数：" + str(self.messageCons)
result = info_0 + "\n" + info_1 + "\n" + info_2 + "\n" + info_4 + "\n" + info_5 + "\n" + info_6 + "\n"
result = result + info_8 + "\n" + info_9 + "\n" + info_10 + "\n" + info_11 + "\n" + info_12 + "\n"
self.writeToFile("./result_pre.txt",result)
time.sleep(5)
timeArray = time.localtime(timeStart / 1000)
testStart = time.strftime("%Y-%m-%d %H:%M:%S", timeArray)
timeEnd = int(time.time() * 1000)
timeArray = time.localtime(timeEnd / 1000)
testestEnd = time.strftime("%Y-%m-%d %H:%M:%S", timeArray)
time.sleep(2) #防止线程慢的时候，某些线程被漏统计的情况
info_0 = "-------------------------- 统计信息 --------------------------"
print(info_0)
totalMsg = 0
for i in self.messageCon:
totalMsg = totalMsg + i
info_1 = "耗时：" + str(timeExpend) + " 毫秒产生了" + str(self.threadCount) + "线程"
info_2 = "开始测试时间：" + testStart
info_3 = "结束测试时间：" + testestEnd
info_4 = "设置socket超时时间：" + str(self.timeOut)
info_5 = "设置线程持续时间：" + str(self.durTime)
info_6 = "成功线程数：" + str(self.sucessNum)
info_7 = "消息总数：" + str(totalMsg)
info_8 = "连接失败：" + str(self.connectTimeoutNum)
info_9 = "发送失败：" + str(self.sendTimeoutNum)
info_10 = "接收失败：" + str(self.reviceTimeoutNum)
print(info_1)
print(info_2)
print(info_3)
print(info_4)
print(info_5)
print(info_6)
print(info_7)
print(info_8)
print(info_9)
print(info_10)
self.writeToFile("../../data/threadDetailsContinuous.json",json.dumps(self.threadArr))
info_11 = self.getInfoFromResult("../../data/threadDetailsContinuous.json")
result = info_0 + "\n" + info_1 + "\n" + info_2 + "\n" + info_3 + "\n" + info_4 + "\n" + info_5 + "\n" + info_6 + "\n"
result = result + info_7 + "\n" + info_8 + "\n" + info_9 + "\n" + info_10 + "\n" + info_11
self.writeToFile("./result.txt", result)
def writeToFile(self,path,data):
with open(path, "w", encoding='utf-8') as fi:
json.dump(data, fi)
# fi.write(data)
if __name__ == "__main__":
t = SendMultMsgThread()
# t.setMsg("7e0002000001314620111800065b7e")
# t.setMsg("7e020001020131462011190001fffc7fff001c010401c0a6380659ad7a02090042003b200204185704310102EA6600010400000000000204001e7c1f0003050A0001f400000405020001d4c000050400057d0240000604000119400007040007530000100c0004006403f203f203f203f2001114ffffffffffffffffffff00200000000000000000001202002400130106001D0101EB7960C0020bb860D0013c62f00203216050014c60F0015860B001146330011c646001416490012060A00201146014010160100102610002022661100201f561F0020e746210040000119c6040012c60700200e660E00203206701010067020100670301016704024e20670502000067060200416707040000017d02097e")
t.startThread()
#json.dump(data, fi)
fi.write(data)
#获取随机消息数据(M300车机)
def getRandomMsg_M300(self,carId):
# carId = 201912010002
wh = random.randint(0,2)
msg = ""
if wh == 0:
hearbeat_msg = "7e000400e14d" + str(carId) + "0000"
hearbeat_msg = hearbeat_msg + self.getCheckCode(hearbeat_msg[2:]) + "7e"
hearbeat_msg = "7e" + self.replace7e7d(hearbeat_msg[2:][:-2]) + "7e"
msg = hearbeat_msg
elif wh == 1:
GPS_msg = "7e002000fb4d" + str(carId) + "002414031003351501c3422106588b8c0703200c520000000000007403200100000000000000"
GPS_msg = GPS_msg + self.getCheckCode(GPS_msg[2:]) + "7e"
GPS_msg = "7e" + self.replace7e7d(GPS_msg[2:][:-2]) + "7e"
msg = GPS_msg
elif wh == 2:
OBD_msg = "7e000300fa4d" + str(carId) + "005d140310033513117f3f0effff3f30f9000001000000000000000000fffffe8000740200008e8800002ac2010003e8503232320000501403e80069006903840000000000640a03e802580000000000000384000000000000000000000000"
OBD_msg = OBD_msg + self.getCheckCode(OBD_msg[2:]) + "7e"
OBD_msg = "7e" + self.replace7e7d(OBD_msg[2:][:-2]) + "7e"
msg = OBD_msg
return msg
def getInfoFromResult(self,path):
with open(path, "r", encoding='utf-8') as fi:
d = fi.read()
d = json.loads(d)
msgC = 0
failT = 0
for i in d:
msgC = msgC + d[i]["msgCon"]
if d[i]["status"] == 1:
failT = failT +1
info_1 = "文件统计总消息数：" + str(msgC)
info_2 = "文件统计总失败数：" + str(failT)
print(info_1)
print(info_2)
return info_1 + "\n" +info_2 + "\n"
#######################################################
# 获取校验码（M300,新硬件）
#######################################################
def getCheckCode(self,data="aa"):
if len(data) % 2 == 1:
raise RuntimeError('数据段错误！')
start = data[0:2]
tmp = int(start,16)
for i in range(2,len(data),2):
tmp = tmp ^ int(data[i:i + 2],16)
dataHex = self.int2hexStringByBytes(tmp)
return dataHex
#######################################################
# 替换消息中的7e7d字符
#######################################################
def replace7e7d(self,data):
tmpR = data
tmp = tmpR[0:2]
tmpA = tmpR[0:2]
tmpR = tmpR[2:]
data = ""
while tmpA != "":
if tmp == "7d":
tmp = "7d01"
elif tmp == "7e":
tmp = "7d02"
data = data + tmp
tmp = tmpR[0:2]
tmpA = tmpR[0:2]
tmpR = tmpR[2:]
return data
#####################################################
# 数字转换为16进制字符串，通过传入字节数可自动补0
# 传入数据格式所占字节数
#####################################################
def int2hexStringByBytes(self, num,bytescount=1):
hexStr = hex(num)[2:]
while len(hexStr) < (bytescount * 2):
hexStr = "0" + hexStr
return hexStr
if __name__ == "__main__":
t = SendMultMsgThread()
# t.setHost("10.100.12.32")
t.setHost("10.100.5.251")
# t.setPort(9008) #M500
t.setPort(9009) #M300
# t.startThread()
t.startThreadContinuous()
# SendMultMsgThread().getInfoFromResult("../../data/threadDetailsContinuous.json")
\ No newline at end of file
lib/multiThread/SendMultMsgThread_m500.py
View file @ f0bb02f9
......@@ -18,12 +18,12 @@ class SendMultMsgThread():
self.msg = msg
self.timeOut = 30 #socket超时时间
self.BUF_SIZE = 1024 #接收消息缓存
self.threadCount = 7000 #并发线程数
self.threadCount = 1000 #并发线程数
self.totalTime = 0 #所有线程的运行总和
self.threadArr = {} #保存每个线程的信息
self.failThreadCount = 0 #失败线程数
self.durThreads = [] #持续发送线程数组，当数组为空，表示所有线程已经结束
dt = 1 * 30 * 60
dt = 1 * 1 * 60
self.durTime = dt #线程持续时间
self.connectTimeoutNum = 0 #连接超时线程数
self.sendTimeoutNum = 0 #发送超时线程数
......@@ -289,33 +289,17 @@ class SendMultMsgThread():
msg = OBD_msg
return msg
#获取随机消息数据(M300车机)
def getRandomMsg_M300(self,carId):
# carId = 201912010002
wh = random.randint(0,2)
msg = ""
if wh == 0:
hearbeat_msg = "4040000e00044d" + str(carId) + "8000000300cf91"
hearbeat_msg = hearbeat_msg[:-4] + self.crc16(hearbeat_msg[:-4])
msg = hearbeat_msg
elif wh == 1:
GPS_msg = "4040003d00054d" + str(carId) + "001001140305031e0301c329ed0659dec501f402e8000000b4050a0b0c9305050258001400000fa0000000005e606f115e60723be44b"
GPS_msg = GPS_msg[:-4] + self.crc16(GPS_msg[:-4])
msg = GPS_msg
elif wh == 2:
OBD_msg = "4040007000064d" + str(carId) + "00120114030503202d26d7fffff0000000000505000000143c00000bb80100000fa00000000a0000000000005e60723b723b39331e100055320000001312001007d0001e0000000000000096000000280096ffff3e0001f40000003e00000000000000000000007213"
OBD_msg = OBD_msg[:-4] + self.crc16(OBD_msg[:-4])
msg = OBD_msg
return msg
# 获取随机消息数据(新硬件车机)
def getRandomMsg_new(self, carId):
# carId = 201912010002
wh = random.randint(0, 2)
msg = ""
wh = 0
if wh == 0:
hearbeat_msg = "4040000e00044d" + str(carId) + "8000000300cf91"
hearbeat_msg = hearbeat_msg[:-4] + self.crc16(hearbeat_msg[:-4])
hearbeat_msg = hearbeat_msg + self.getCheckCode(hearbeat_msg[2:]) + "7e"
hearbeat_msg = self.replace7e7d(hearbeat_msg)
msg = hearbeat_msg
elif wh == 1:
GPS_msg = "4040003d00054d" + str(
......@@ -330,7 +314,7 @@ class SendMultMsgThread():
return msg
####################################################
# 定义生成校验字段的函数
# 定义生成校验字段的函数（M500 校验方式）
# inputStr:需要传入一个已经转换为16进制的字符串
#####################################################
# add crc 16 check at the end of the string
......@@ -358,7 +342,6 @@ class SendMultMsgThread():
return outputStr
else:
return inputStr
# pad zero to the right of the string if not long enough
def rightPad(self,inputStr, strLen):
if (strLen > len(inputStr)):
......@@ -385,14 +368,13 @@ class SendMultMsgThread():
return info_1 + "\n" +info_2 + "\n"
if __name__ == "__main__":
t = SendMultMsgThread()
# t.setHost("10.100.12.32")
t.setHost("10.100.5.251")
t.setPort(9008)
t.setPort(9008) #M500
# t.setPort(9009) # M300
# t.startThread()
t.startThreadContinuous()
......
lib/socket/messageTest.py
View file @ f0bb02f9
......@@ -22,14 +22,14 @@ port = 9001
# msg = MessageBase().generateMsg()
# msg = TerminalCommonMsgRes_msg().generateMsg() #终端通用应答
# msg = TerminalHeartbeat_msg().generateMsg() #终端心跳
msg = TerminalHeartbeat_msg().generateMsg() #终端心跳
# msg = TerminalRegister_msg().generateMsg() #终端注册
# msg = TerminalCancle_msg().generateMsg() #终端注销
# msg = TerminalAuthenticate_msg().generateMsg() #终端鉴权
# msg = TerminalVersionInfo_msg().generateMsg() #终端版本信息上报
# msg = QueryTerminalParam_res().generateMsg() #查询终端参数应答
# msg = QueryTerminalProperty_res().generateMsg() #查询终端属性应答消息
msg = Location_msg().generateMsg() #位置信息汇报
# msg = Location_msg().generateMsg() #位置信息汇报
# msg = DataUpstreamTransport_msg().generateMsg() #数据上行透传消息
# msg = TerminalUpdataResult_msg().generateMsg() #终端升级结果通知
# msg = LocationDataBatchUpdate_msg().generateMsg() #定位数据批量上传
......
lib/socket/protocolTest_M300.py 0 → 100644
View file @ f0bb02f9
#coding:utf-8
import binascii
import socket
from lib.protocol.report.GPSReport_protocol import GPSReport_protocol
from lib.protocol.report.OBDReport_CAN_protocol import OBDReport_CAN_protocol
from lib.protocol.report.OBDReport_CAN_protocol import OBDReport_CAN_protocol
from lib.protocol.report.HeartBeatReport_protocol import HeartBeatReport_protocol
from lib.protocol.report.LoginReport_protocol import LoginReport_protocol
from lib.protocol.report.SecurityStatusReport_protocol import SecurityStatusReport_protocol
from lib.protocol.report.BaseStationReport_protocol import BaseStationReport_protocol
from lib.protocol.report.TroubleReport_protocol import TroubleReport_protocol
from lib.protocol.report.EventReport_protocol import EventReport_protocol
# host = "10.100.12.34"
# port = 8712
# host = "10.100.12.32"
# port = 9009
# host = "10.100.5.251"
# port = 9009
host = "10.100.5.251"
port = 9009
msg = "7e000400e14d2019120315000000957e" #终端上报心跳协议
print(msg)
BUF_SIZE = 1024
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) # 在客户端开启心跳
client.connect((host, port))
client.send(binascii.a2b_hex(msg))
# client.send(bytes.fromhex(msg))
data = client.recv(BUF_SIZE)
print(data)
print(binascii.b2a_hex(data))
client.close()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment