引言
在AK公司工作的时候，遇到一个设备绑定的问题。 专项测试的内容是，将IPC绑定到一个基站设备上面，对这个绑定过程进行压测，统计成功率。要求成功率是99.9%以上。
该专项项目没有通过，分析其原因卡在IPC和基站的TCP连接过程，发现基站那边很久没有给到响应。
当时用电脑模拟IPC，做了一个TCP连接的压测工具。
正文
import socket
import time
import datetime
import os
from logzero import logger, logfile
def get_timestamp_str():
''' 获取当前时间戳，到秒 '''
return datetime.datetime.now().strftime('%y%m%d_%H%M%S')
# ---------------- 初始化日志相关 ------------------------------
RUN_TIMESTAMP = get_timestamp_str()
if not os.path.exists('./log/'):
os.mkdir('./log')
new_log_file = f'./log/test2_{RUN_TIMESTAMP}.log'
# 固定的IP地址和端口
IP_ADDRESS = "192.168.32.2" # 明确基站的IP地址是固定的IP地址，如果不是，需要手动修改
PORT = 10400
# 重试间隔时间（秒）
RETRY_INTERVAL = 4
INIT_TIMEOUT = 2
MAX_TIMEOUT = 12
def attempt_connection(seq, timeout):
"""
尝试建立TCP连接
:param seq: 当前的序号
:param timeout: 当前的超时时间
:return: 是否连接成功
"""
try:
# 创建一个socket对象
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(timeout) # 设置超时时间
# 尝试连接
s.connect((IP_ADDRESS, PORT))
local_ip, local_port = s.getsockname()
logger.info(f"轮次:{seq}，本次建连成功，超时参数：{timeout} 秒, 本地端口: {local_port}")
return True
except socket.timeout:
local_ip, local_port = s.getsockname()
logger.warning(f"轮次:{seq}，连接超时，当前超时参数：{timeout} 秒, 本地端口: {local_port}")
except Exception as e:
if s:
local_ip, local_port = s.getsockname()
logger.error(f"轮次:{seq}，连接失败，超时参数: {timeout} 秒, 错误信息：{e}, 本地端口: {local_port}")
else:
logger.error(f"轮次:{seq}，连接失败，超时参数: {timeout} 秒, 错误信息：{e}, 本地端口: 空")
return False
def main():
# 打印参数
logger.info(f"TCP连接测试, IP地址: {IP_ADDRESS}, 端口: {PORT}, 重试间隔: {RETRY_INTERVAL}秒, 初始超时时间: {INIT_TIMEOUT} 秒, 最大超时时间: {MAX_TIMEOUT}秒")
logger.info('-----------------------------------------------------------------------------------------------------------------------')
seq = 1
while True:
timeout = INIT_TIMEOUT # 初始超时时间
while timeout <= MAX_TIMEOUT:
if attempt_connection(seq, timeout):
break # 如果连接成功，退出当前循环
timeout += 1 # 超时时间加1
time.sleep(2)
else:
logger.info(f"尝试 {MAX_TIMEOUT} 秒超时后仍然失败，等待下轮重试")
seq += 1
time.sleep(RETRY_INTERVAL) # 等待RETRY_INTERVAL时间后再次尝试
if __name__ == "__main__":
main()
总结
这个测试工具随时可以ctrl+c停止下来，相关日志会记录已经测试过的轮数。
参考链接