一、使用
1、代码
import argparse
import socket
import time
import subprocess
import re
import sys
def measure_latency(host):
# 使用ping命令测量延迟
try:
# 根据操作系统选择ping参数
if sys.platform.startswith('win'):
output = subprocess.check_output(['ping', '-n', '4', host], stderr=subprocess.STDOUT, text=True)
match = re.search(r'平均 = (\d+)ms', output)
else:
output = subprocess.check_output(['ping', '-c', '4', host], stderr=subprocess.STDOUT, text=True)
match = re.search(r'avg/(\d+\.\d+)', output) # Linux ping输出格式不同
if match:
return int(match.group(1))
else:
return None
except subprocess.CalledProcessError:
return None
def server_mode(port):
# 创建TCP服务器
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('0.0.0.0', port))
s.listen()
print(f"服务器监听在端口 {port}...")
conn, addr = s.accept()
with conn:
print(f"连接来自 {addr}")
start_time = time.time()
data_received = 0
while True:
data = conn.recv(1024*1024) # 1MB缓冲区
if not data:
break
data_received += len(data)
end_time = time.time()
duration = end_time - start_time
speed_mbps = (data_received * 8) / (duration * 1024 * 1024) # 转换为Mbps
print(f"接收数据: {data_received / (1024*1024):.2f} MB")
print(f"传输时间: {duration:.2f} 秒")
print(f"速度: {speed_mbps:.2f} Mbps")
def client_mode(host, port, size_mb=100):
# 创建TCP客户端
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
start_connect = time.time()
s.connect((host, port))
connect_time = time.time() - start_connect
print(f"连接建立时间: {connect_time*1000:.2f} ms")
# 准备发送的数据 (随机字节)
data = b'x' * (1024*1024) # 1MB数据块
total_size = size_mb * 1024*1024 # 总大小
blocks = total_size // len(data)
start_time = time.time()
for _ in range(blocks):
s.sendall(data)
end_time = time.time()
duration = end_time - start_time
speed_mbps = (total_size * 8) / (duration * 1024 * 1024)
print(f"发送数据: {size_mb} MB")
print(f"传输时间: {duration:.2f} 秒")
print(f"速度: {speed_mbps:.2f} Mbps")
def main():
parser = argparse.ArgumentParser(description='内网网速和延迟测试工具')
parser.add_argument('-s', '--server', action='store_true', help='运行服务器模式')
parser.add_argument('-c', '--client', action='store_true', help='运行客户端模式')
parser.add_argument('-i', '--host', help='服务器IP地址 (客户端模式)')
parser.add_argument('-p', '--port', type=int, default=5000, help='端口号 (默认5000)')
parser.add_argument('-size', '--size', type=int, default=100, help='测试文件大小(MB) (默认100MB)')
args = parser.parse_args()
if args.server:
server_mode(args.port)
elif args.client:
if not args.host:
print("客户端模式需要指定服务器IP地址 (-i)")
return
# 测量延迟
latency = measure_latency(args.host)
if latency is not None:
print(f"延迟: {latency} ms")
else:
print("无法测量延迟")
# 测试网速
client_mode(args.host, args.port, args.size)
else:
print("请指定服务器模式 (-s) 或客户端模式 (-c)")
if __name__ == '__main__':
main()
2、使用
# 服务端
python netspeedtest.py -s -p 5000
# 客户端
python netspeedtest.py -c -i 192.168.1.100 -p 5000 -size 200
3、注意事项
1、两台服务器之间网络通畅
2、端口能通,防火墙是关闭的
3、需要python环境
4、由于网络抖动,可以多次测试求平均值。