本文档是 iperf3 图形化工具 的官方完整说明书，适用于各个阶层、各个行业、不同技术背景的用户。无论您是网络管理员、系统工程师、学生、教师、网络爱好者，还是只是想测试家庭网络性能的普通用户，本文档都能为您提供所需的指导。
- 作者：杜玛 (Duma)
- 版权：永久保留所有权利
- 项目地址：https://github.com/duma520
- 问题报告：通过 GitHub Issues 提交
- 特别说明：我们不提供私人邮箱支持，所有技术支持都通过公开渠道进行，以便其他用户也能受益。
- 多层次内容：从零基础入门到高级专业配置
- 多角度解读：兼顾理论知识和实践操作
- 多示例说明：丰富的实际应用场景示例
- 全功能覆盖：详细解释每个选项和功能
- 问题导向：针对常见问题提供解决方案
iperf3 是一个专业的网络性能测试工具，用于测量两个节点之间的最大 TCP/UDP 带宽性能。它是网络工程师、系统管理员和开发人员必备的工具之一。
通俗解释：就像用测速软件测试您的网速一样，iperf3 是专业的"网络测速仪"，但功能更强大、更精确。
原始的 iperf3 是命令行工具，需要记住复杂的参数。本图形化工具提供了：
- 直观的界面：无需记忆命令参数
- 完整的参数支持：支持所有 iperf3 选项
- 结果可视化：直观显示测试结果
- 配置管理：保存和加载测试配置
- 历史记录：追踪历次测试结果
- 普通家庭用户：测试家庭宽带速度、Wi-Fi 信号强度
- 网络爱好者：学习网络知识、测试网络设备性能
- IT 管理员：诊断网络问题、监控网络性能
- 系统工程师：容量规划、性能优化
- 开发人员：测试应用程序的网络性能
- 学生/教师：网络课程教学、实验
- Python 3.6+ （如果显示版本错误，请升级Python）
- iperf3 程序 （这是实际执行测试的核心工具）
Windows 用户：
- 安装 Python：从 python.org 下载并安装
- 安装 iperf3：从 iperf.fr 下载 Windows 版本
- 将 iperf3.exe 放在系统 PATH 或程序所在目录
Linux/macOS 用户：
# Ubuntu/Debian
sudo apt-get install python3 python3-pip iperf3
# CentOS/RHEL
sudo yum install python3 python3-pip iperf3
# macOS
brew install python iperf3- 双击运行：
iperf_gui.py或使用命令行python iperf_gui.py - 界面介绍：
- 顶部标签页：不同功能区域
- 中间区域：参数配置
- 底部按钮：开始/停止测试
- 右侧结果区：显示测试结果
- 模式选择："客户端"
- 服务器地址输入：
127.0.0.1（这是本机地址） - 点击"开始测试"
这是什么？ 测试您自己电脑的网络栈性能，确保工具正常工作。
准备两台电脑（A 和 B）：
-
在电脑 B（服务器）：
- 模式选择："服务器"
- 点击"开始测试"
- 记下 B 的 IP 地址（如 192.168.1.100）
-
在电脑 A（客户端）：
- 模式选择："客户端"
- 服务器地址输入：
192.168.1.100（B 的 IP） - 点击"开始测试"
结果解读：显示的是 A 到 B 的网络速度。如果您的网络是千兆（1Gbps），理想值应该是 940Mbps 左右（理论值的 94%）。
这是最常用的配置区域，包含网络测试的基本参数。
- 客户端：主动发起测试的一端
- 适用场景：测试从本地到服务器的速度
- 举例：测试从办公室电脑到公司服务器的网速
- 服务器：等待连接的一端
- 适用场景：让他人测试连接到您的速度
- 举例：您的朋友想测试从他家到您家的网络速度
专业说明：
- 客户端模式对应
iperf3 -c命令 - 服务器模式对应
iperf3 -s命令 - 一次完整的测试需要一端是服务器，另一端是客户端
- 格式：IP 地址或域名
- 示例：
192.168.1.1（局域网 IP）10.0.0.1（内网 IP）www.example.com（域名，会自动解析为 IP）127.0.0.1（本地回环地址，用于测试本机）
重要提示：
- 如果服务器在 NAT 后面（如家庭路由器），需要端口转发
- 确保防火墙允许 5201 端口（或您指定的端口）
- 默认值：5201
- 可修改范围：1-65535
- 为什么改端口？
- 安全原因：避免使用默认端口
- 多实例运行：同时运行多个测试
- 端口冲突：5201 端口已被占用
示例场景：
- 家庭宽带测试：使用默认 5201
- 企业环境：使用 15201 等非标准端口
- 同时测试：A用5201，B用5202，C用5203...
- TCP：传输控制协议
- 特点：可靠、有序、错误重传
- 适用：网页浏览、文件下载、视频流
- 测试内容：实际可用带宽、延迟影响
- UDP：用户数据报协议
- 特点：快速、无连接、可能丢包
- 适用：实时视频、语音通话、在线游戏
- 测试内容：最大吞吐量、丢包率、抖动
专业对比：
| 特性 | TCP | UDP |
|---|---|---|
| 可靠性 | 高（自动重传） | 低（不保证到达） |
| 速度 | 相对较慢 | 非常快 |
| 顺序性 | 保证顺序 | 不保证顺序 |
| 适用场景 | 文件传输、网页 | 实时应用、游戏 |
| 测试重点 | 实际带宽 | 最大带宽、质量 |
- 单位：秒
- 范围：1-3600 秒（1秒到1小时）
- 建议值：
- 快速测试：10-30 秒
- 稳定性测试：300-600 秒（5-10分钟）
- 压力测试：3600 秒（1小时）
时间选择策略：
-
短时间测试（10秒）：快速检查
- 优点：快速得到结果
- 缺点：可能受瞬时波动影响
- 适用：日常检查、故障初步排查
-
中时间测试（60秒）：标准测试
- 优点：结果相对稳定
- 缺点：需要等待
- 适用：网络验收、性能评估
-
长时间测试（300秒以上）：稳定性测试
- 优点：反映真实稳定性能
- 缺点：耗时较长
- 适用：链路稳定性评估、QoS验证
专业建议：
- 对于波动较大的网络（如无线），建议测试时间 ≥ 30秒
- 对于需要精确测量的场景（如SLA验证），建议 ≥ 300秒
- TCP测试需要足够时间建立稳定传输速率
- 是什么：同时建立多个连接进行测试
- 范围：1-128 个流
- 默认值：1
- 单位：个连接
通俗解释： 想象一条公路：
- 1个流 = 1条车道
- 4个流 = 4条车道同时通行
- 结果 = 总通行能力
实际应用场景：
-
单流（默认）：
- 测试单连接性能
- 模拟普通文件下载
- 结果反映单连接最大速度
-
多流（建议4-8）：
- 测试多用户同时访问
- 模拟实际应用场景（如视频会议+文件传输）
- 更能反映真实网络性能
-
大量流（16-32）：
- 压力测试
- 测试网络设备并发处理能力
- 专业性能评估
技术原理：
单流：客户端 ------------ 服务器
多流：客户端 ---流1--- 服务器
---流2---
---流3---
示例设置建议：
- 家庭网络测试：1-4个流
- 企业网络评估：8-16个流
- 数据中心测试：16-32个流
- 极限压力测试：32-128个流
注意事项：
-
过多的流可能导致：
- 路由器/交换机CPU负载过高
- 测试结果反而下降
- 影响其他网络应用
-
流数量与测试结果的关系：
- 初期：增加流数量，总带宽增加
- 最佳点：达到最大带宽
- 后期：增加流，带宽不再增加甚至下降
-
发现最佳流数量： 推荐使用"阶梯测试法"：
第1次：1个流 → 记录结果 第2次：4个流 → 记录结果 第3次：8个流 → 记录结果 第4次：16个流 → 记录结果 比较结果，找到最佳值
- 功能：人为限制测试使用的带宽
- 格式：数字+单位，如
100M、1G - 单位：
- K = Kbps（千比特每秒）
- M = Mbps（兆比特每秒）
- G = Gbps（千兆比特每秒）
为什么需要限制带宽？
-
避免网络拥塞：
- 测试时不希望影响其他业务
- 家庭中不影响家人上网
- 企业中不影响生产系统
-
模拟特定场景：
- 测试在100M限制下的应用性能
- 模拟低速链路（如移动网络）
- 验证QoS策略效果
-
渐进式测试：
- 从低带宽开始，逐步增加
- 观察网络在不同负载下的表现
应用示例：
场景1：家庭宽带是500M，但只想测试100M
设置：带宽限制 = 100M
场景2：测试4K视频流所需带宽
设置：带宽限制 = 25M （典型4K视频码率）
场景3：模拟3G网络环境
设置：带宽限制 = 5M （典型3G速度）
专业技巧：
-
发现实际带宽：
- 不设置限制，测试得到最大带宽
- 然后设置为最大带宽的80%进行稳定性测试
-
测试缓冲区影响：
步骤1：带宽限制 = 100M，测试 步骤2：带宽限制 = 200M，测试 步骤3：带宽限制 = 500M，测试 观察不同限制下的性能变化 -
验证网络设备能力：
- 设置接近设备标称值
- 观察是否能稳定达到
- 测试长时间稳定性
注意事项：
-
实际限制可能受以下因素影响：
- 网络设备处理能力
- 两端电脑性能
- 其他网络流量
-
如果设置限制但达不到：
- 可能是网络本身达不到该速度
- 可能是其他瓶颈（如磁盘IO、CPU）
-
与"服务器比特率限制"的区别：
- 客户端限制：控制发送速率
- 服务器限制：控制接收速率
- 通常只需设置一端
- 技术名称：TCP Window Size
- 作用：控制"在途数据"的最大量
- 单位：字节，通常用K、M表示
- 默认值：系统自动调整
通俗比喻： 想象一个送货流程：
- 货物 = 数据包
- 货车容量 = 窗口大小
- 装满一车送一次 = 发送窗口数据
- 更大的货车 = 更高的效率（在一定范围内）
详细解释：
-
什么是TCP窗口？ TCP使用滑动窗口机制控制流量：
发送方： [已确认][发送窗口][未发送] ↑ 窗口大小窗口大小决定了一次可以发送多少数据而不需要等待确认。
-
为什么重要？ 公式：
最大吞吐量 = 窗口大小 / 往返时间(RTT)例子：
- RTT = 50ms（典型局域网）
- 窗口大小 = 64KB
- 最大吞吐量 = 64KB / 0.05s = 10.5Mbps
结论：对于高延迟网络，需要更大的窗口才能达到高速。
-
如何选择合适的窗口大小？
公式：窗口大小 ≥ 带宽 × RTT 示例计算： 带宽目标：1Gbps = 125MB/s RTT：20ms = 0.02s 所需窗口大小 = 125MB/s × 0.02s = 2.5MB
应用场景：
-
局域网测试：
- RTT通常 < 1ms
- 默认窗口通常足够
- 可以测试64K、128K、256K对比
-
广域网测试：
- RTT可能 > 50ms
- 需要增大窗口
- 建议：1M、2M、4M、8M 测试
-
互联网测试：
- RTT可能 > 100ms
- 窗口需要更大
- 建议：2M、4M、8M、16M
设置建议：
网络类型 典型RTT 建议窗口大小
局域网 <1ms 64K-256K
城域网 5-20ms 512K-2M
国内跨省 30-80ms 2M-8M
国际链路 100-300ms 8M-32M
卫星链路 >500ms >64M
专业调试步骤：
- 不设置窗口（使用默认），测试得到基准
- 逐步增大窗口，观察吞吐量变化：
测试1：窗口 = 64K 测试2：窗口 = 128K 测试3：窗口 = 256K 测试4：窗口 = 512K ... 直到吞吐量不再增加 - 记录最佳窗口值
注意事项：
-
窗口不是越大越好：
- 过大窗口可能导致拥塞
- 消耗更多内存
- 丢包时重传更多数据
-
需要考虑接收方窗口： TCP通信中，实际窗口 = min(发送方窗口, 接收方窗口) 两端都需要适当设置。
-
与MSS的关系：
- MSS是单个包的大小
- 窗口是多个包的总量
- 通常：窗口大小是MSS的整数倍
实际案例： 案例：公司总部（北京）到分公司（上海）专线
- 带宽：100Mbps
- 测量RTT：35ms
- 计算理想窗口：100Mbps × 0.035s ≈ 3.5Mbits = 437KB
- 测试设置：从256K开始，逐步增加到1M、2M、4M
- 结果：2M窗口时达到最佳性能
- 替代选项：与"测试时间"二选一
- 格式：数字+单位，如
100M、1G - 单位：
- K = KB（千字节）
- M = MB（兆字节）
- G = GB（千兆字节）
与"测试时间"的区别：
- 测试时间：固定时长，看能传多少数据
- 传输数据量：固定数据量，看需要多少时间
适用场景：
-
测试特定文件传输：
- 模拟传输1GB文件需要的时间
- 预估大文件传输耗时
-
短时突发测试：
- 只传输100MB数据
- 快速测试，避免长时间占用网络
-
一致性测试：
- 每次都传输相同数据量
- 比较不同时间/配置下的性能
示例：
场景：测试备份1TB数据到云存储的时间
设置：传输数据量 = 1G （先用小量测试）
结果：传输1G需要30秒
推算：1TB ≈ 1000 × 30秒 = 8.3小时
- 作用：指定传输的块（block）数量
- 单位：个
- 适用：UDP测试或特定测试场景
技术细节：
- 每个块包含多个数据包
- 对于UDP，可以控制发送的数据包数量
- 对于TCP，较少使用此参数
应用场景：
-
UDP包数量测试：
- 发送10000个UDP包
- 统计丢包率
- 测试网络设备包转发能力
-
特定协议模拟：
- 模拟VoIP通话（每秒50个包）
- 模拟传感器数据（每秒10个包）
- 作用：控制读写缓冲区大小
- 单位：字节，通常用K表示
- 默认：系统默认（通常8K或64K）
缓冲区的作用：
应用程序 → 应用缓冲区 → TCP栈 → 网络
接收方：网络 → TCP栈 → 应用缓冲区 → 应用程序
设置建议：
- 常规测试：使用默认值
- 高性能网络：增大缓冲区（如64K、128K）
- 高延迟网络：需要更大缓冲区
缓冲区与窗口的关系：
- 缓冲区：在内存中暂存数据
- 窗口：在网络中在途数据
- 缓冲区 ≥ 窗口 才能充分发挥性能
- 作用：控制结果输出的频率
- 单位：秒
- 范围：0.1-10.0秒
- 默认：1.0秒
报告内容： 每间隔时间输出一次统计，包括：
- 时间区间
- 传输数据量
- 带宽
- （UDP）抖动和丢包
设置策略：
-
详细监控：0.1-0.5秒
- 观察瞬时波动
- 调试网络问题
- 缺点：输出信息多
-
标准测试：1.0秒（默认）
- 平衡详细度和可读性
- 适合大多数场景
-
长时间测试：2.0-5.0秒
- 减少输出信息
- 观察趋势变化
输出示例：
[ ID] Interval Transfer Bandwidth
[ 5] 0.00-1.00 sec 117 MBytes 982 Mbits/sec
[ 5] 1.00-2.00 sec 118 MBytes 989 Mbits/sec
[ 5] 2.00-3.00 sec 119 MBytes 998 Mbits/sec
- 作用：控制结果显示的单位
- 选项：自动、K、M、G、T
- 默认：自动（根据数值自动选择）
各单位的含义：
- K = Kbits/sec 或 KBytes/sec
- M = Mbits/sec 或 MBytes/sec
- G = Gbits/sec 或 GBytes/sec
- T = Tbits/sec 或 TBytes/sec
选择建议：
- 自动：让程序根据数值大小自动选择
- 固定单位：强制使用特定单位，便于比较
示例：
实际值：952 Mbits/sec
显示为：
- 自动：952 Mbits/sec
- 强制K：975,000 Kbits/sec
- 强制G：0.95 Gbits/sec
- 全称：Maximum Segment Size
- 作用：控制TCP数据包的有效载荷大小
- 单位：字节
- 默认：1460（以太网标准）
通俗解释： MSS就是TCP数据包中实际数据的最大大小（不包括TCP头部和IP头部）。
详细说明：
完整数据包结构：
[以太网头][IP头][TCP头][数据][以太网尾]
↑
MSS
典型值：1500(MTU) - 20(IP头) - 20(TCP头) = 1460
为什么需要调整MSS？
-
网络MTU限制：
- 某些网络MTU不是1500（如PPPoE是1492）
- 需要相应减小MSS
- 公式：MSS = MTU - 40（IP头20+TCP头20）
-
避免分片：
- 数据包过大可能被分片
- 分片降低性能，增加丢包风险
- 适当MSS可避免分片
-
优化性能：
- 某些场景下特定MSS可能更优
- 需要实验测试
常见MTU/MSS组合：
网络类型 MTU 建议MSS
标准以太网 1500 1460
PPPoE 1492 1452
Jumbo Frame 9000 8960
VPN隧道 1400-1420 对应减少
设置建议：
-
一般情况：保持默认0（使用系统默认）
-
遇到问题时：
- 测试标准MSS：1460
- 测试PPPoE MSS：1452
- 测试小包：536（最小安全值）
-
诊断方法：
步骤1：ping测试MTU ping -l 1472 -f 目标IP 如果不通，逐步减小直到通 步骤2：计算MSS MSS = 成功值 + 28 - 40 步骤3：在iperf中设置该MSS
专业应用：
-
VPN环境：
- VPN增加额外头部
- 需要减小MSS
- 通常设置为1350-1400
-
广域网优化：
- 某些设备对特定MSS有优化
- 需要实验确定最佳值
-
TCP性能调优：
- MSS影响窗口缩放
- 影响重传效率
- 专业调优参数
注意事项：
-
两端需要匹配： TCP连接使用双方较小的MSS值 需要两端都适当设置
-
不是越大越好： 过大MSS可能导致：
- 分片
- 增加延迟
- 增加丢包影响
-
与窗口关系： 窗口大小应该是MSS的整数倍 否则会有部分浪费
包含更多专业选项，适合高级用户和特定场景。
- 作用：设置建立连接的最大等待时间
- 单位：毫秒(ms)
- 默认：0（使用系统默认，通常30-60秒）
适用场景：
- 快速失败：网络不通时快速返回错误
- 严格测试：要求连接必须在指定时间内建立
- 自动化测试：避免长时间等待
设置建议：
网络类型 建议超时
局域网 1000-5000 ms
广域网 5000-30000 ms
高延迟网络 30000-60000 ms
移动网络 10000-30000 ms
- 作用：指定使用哪个网络接口
- 格式：接口名称
- 示例：
eth0、wlan0、en0、以太网、Wi-Fi
为什么需要绑定设备？
- 多网卡选择：电脑有多个网卡时
- 指定路径：强制使用有线或无线
- 测试特定接口：测试某个网卡性能
查看设备名称：
Windows: ipconfig /all
Linux: ip addr 或 ifconfig
macOS: ifconfig 或 networksetup -listallhardwareports
- 作用：指定源IP地址
- 格式：IP地址
- 用途：
- 多IP主机指定源地址
- VPN环境指定出口
- 测试特定源地址的路由
- 作用：指定客户端使用的端口号
- 格式：端口号或范围
- 示例：
60000：固定端口60000-60010：端口范围
应用场景：
- 防火墙规则：只允许特定端口
- NAT穿透：配合端口转发
- 多客户端识别：不同客户端用不同端口
- 作用：跳过测试开始的一段时间
- 单位：秒
- 默认：0（不省略）
为什么需要省略？ TCP需要时间达到稳定状态：
- 慢启动：TCP从低速逐渐增加
- 缓冲区填充：需要时间填满管道
- 避免初始波动影响结果
建议值：
- 短测试（10秒）：省略2秒
- 标准测试（60秒）：省略5秒
- 长测试（300秒+）：省略10秒
- 作用：为输出行添加前缀
- 用途：
- 区分多个同时运行的测试
- 记录测试信息
- 自动化测试标签
示例：
设置标题：办公室到机房
输出：[办公室到机房] [ 5] 0.00-1.00 sec 117 MBytes 982 Mbits/sec
- 作用：在JSON输出中包含自定义数据
- 格式：字符串
- 用途：记录测试环境信息
示例：
额外数据：location=beijing;device=routerA;test_id=20231201_001
- 选项：自动、仅IPv4、仅IPv6
- 默认：自动
选择策略：
- 自动：让系统自动选择（推荐）
- 仅IPv4：强制使用IPv4
- 仅IPv6：强制使用IPv6
应用场景：
- IPv6测试：选择仅IPv6
- 兼容性测试：分别测试IPv4和IPv6
- 问题诊断：确定是IPv4还是IPv6的问题
- 作用：设置IP包的服务类型字段
- 范围：0-255
- 默认：0
常见TOS值：
0x00 (0): 一般服务
0x10 (16): 最小延迟（如SSH、Telnet）
0x08 (8): 最大吞吐量（如FTP）
0x04 (4): 最高可靠性
0x02 (2): 最小成本
实际应用：
- 测试QoS：验证不同TOS值的优先级
- 模拟应用：模拟VoIP（最小延迟）或FTP（最大吞吐量）
- 网络调优：测试网络设备对TOS的处理
- 作用：更精细的QoS标记
- 范围：0-63
- 默认：0
常见DCSD值：
CS0 (0): 默认
EF (46): 加速转发（语音）
AF41 (34): 保证转发（视频）
AF31 (26): 保证转发（语音信令）
AF21 (18): 保证转发（交互式）
AF11 (10): 保证转发（批量数据）
- 作用：控制数据发送的节奏
- 单位：微秒(μs)
- 默认：根据带宽自动计算
技术原理： 定时器控制发送数据包的时间间隔：
- 值越小：发送越密集，可能突发
- 值越大：发送越平稳，更均匀
计算公式：
定时器 ≈ 包大小 / 目标带宽
示例：目标1Gbps，包1500字节
定时器 ≈ 1500×8 bits / 1e9 bps ≈ 12 μs
- 作用：设置接收数据的超时时间
- 单位：毫秒(ms)
- 默认：0（不超时）
应用场景：
- 不稳定网络：网络频繁中断
- 移动网络：信号可能丢失
- 严格测试：要求持续稳定传输
- 作用：交换发送和接收方向
- 命令：
-R
通俗解释： 正常情况下：客户端发送，服务器接收 反向测试：服务器发送，客户端接收
应用场景：
- 测试上行带宽：从服务器到客户端
- 非对称链路测试：如ADSL（上行小，下行大）
- 双向测试准备：先测A→B，再测B→A
- 作用：同时进行双向传输
- 命令：
--bidir
与反向测试的区别：
- 反向测试：单向，但方向相反
- 双向测试：同时两个方向
应用场景：
- 全双工测试：测试设备同时收发能力
- 实际应用模拟：如视频会议（双方都在发视频）
- 网络压力测试：双向满负载
- 作用：关闭TCP的Nagle算法
- 命令：
-N
Nagle算法是什么？
- 目的：减少小包数量
- 原理：积累小数据，合并发送
- 优点：减少包数量，提高效率
- 缺点：增加延迟
什么时候禁用？
- 实时应用测试：如游戏、VoIP
- 低延迟需求：需要最小化延迟
- 交互式应用：如Telnet、SSH
注意：禁用Nagle可能增加包数量，降低效率。
- 作用：设置IP包的Don't Fragment标志
- 命令：
--dont-fragment
DF标志的作用：
- 设置后：路由器不能分片此包
- 如果包太大：路由器丢弃并返回错误
- 用途：发现路径MTU
应用场景：
- MTU发现：配合ping使用
- 避免分片：分片降低性能，增加丢包
- 特定测试：测试网络对不分片包的处理
- 作用：使用零拷贝技术发送数据
- 命令：
-Z
零拷贝技术：
- 传统：数据从用户空间→内核空间→网卡
- 零拷贝：数据直接从用户空间→网卡
- 优点：减少CPU使用，提高性能
- 要求：系统支持，特定网卡驱动
适用场景：
- 高性能测试：10G+网络
- CPU受限环境：CPU性能是瓶颈
- 专业评估：测试零拷贝效果
- 作用：在接收端使用零拷贝
- 命令：
--skip-rx-copy
注意事项：
- 需要两端都支持
- 可能不稳定
- 主要用于测试和开发
- 作用：使用64位统计计数器（UDP）
- 命令：
--udp-counters-64bit
为什么需要？
- 32位计数器最大4GB
- 高速长时间测试可能溢出
- 64位支持更大计数
- 作用：UDP负载使用重复模式
- 命令：
--repeating-payload
用途：
- 压缩测试：测试网络设备的压缩能力
- 模式识别：测试DPI设备识别能力
- 特定测试：某些测试场景需要
- 作用：客户端获取服务器端的输出
- 命令：
--get-server-output
应用场景：
- 集中监控：客户端查看两端结果
- 自动化测试：一次获取所有数据
- 对比分析：比较客户端和服务器视角
- 作用：服务器处理一个连接后退出
- 命令：
-1
应用场景：
- 自动化脚本：测试完自动退出
- 临时测试：只测一次
- 资源清理：避免忘记停止服务器
- 作用：服务器作为守护进程运行
- 命令：
-D
特点：
- 后台运行
- 输出到系统日志
- 适合长期运行
- 作用：RSA加密使用PKCS1填充
- 命令：
--use-pkcs1-padding
安全相关：
- 默认使用OAEP填充（更安全）
- PKCS1用于兼容旧系统
- 除非必要，否则使用默认
- 作用：限制服务器的发送速率
- 格式：如
100M、1G - 与客户端限制的区别：
- 客户端限制：控制发送速率
- 服务器限制：控制接收速率（在反向测试中是发送速率）
应用场景：
- 限制服务器负载：保护服务器不被压垮
- 模拟低性能服务器：测试客户端在低速服务器下的表现
- 非对称测试：模拟上行带宽小的服务器
- 作用：无数据时自动断开连接的时间
- 单位：秒
- 默认：0（不超时）
用途：
- 资源清理：自动断开空闲连接
- 安全考虑：避免连接一直开放
- 自动化管理：无需手动清理
设置建议：
测试场景 建议超时
临时测试 60-300秒
长期运行 3600秒（1小时）
公开服务器 1800秒（30分钟）
- 作用：服务器运行的最长时间
- 单位：秒
- 默认：0（无限制）
应用场景：
- 定时测试：每天运行2小时
- 活动期间：会议期间限时开放
- 资源控制：避免长期占用
- 作用：允许的客户端服务器时间差
- 单位：秒
- 默认：1.0秒
用途：
- 时间同步检查
- 防止时间不同步导致的统计错误
- 安全考虑
- 作用：客户端认证用户名
- 需要配合：服务器端用户列表
- 作用：服务器私钥文件路径
- 格式：文件路径
- 用途：加密通信
- 作用：客户端公钥文件路径
- 用途：验证服务器
- 作用：授权用户配置文件路径
- 格式：JSON或文本文件
示例用户文件：
{
"users": [
{
"username": "testuser",
"password_hash": "..."
}
]
}- 作用：发送指定文件内容
- 命令：
-F 文件名
特点：
- 发送文件内容作为测试数据
- 不实际传输文件
- 用于特定模式测试
- 作用：将进程ID写入文件
- 用途：
- 方便管理
- 自动化脚本控制
- 监控进程状态
- 作用：将输出写入文件
- 用途：
- 长期记录
- 后续分析
- 审计追踪
- 作用：绑定到特定CPU核心
- 格式：CPU编号，如
0、0,1
应用场景：
- 性能测试：避免CPU迁移影响
- 多核优化：绑定到特定核心
- NUMA系统：优化内存访问
- 作用：自定义时间戳格式
- 格式：strftime格式
- 示例：
%H:%M:%S、%Y-%m-%d %H:%M:%S
- 作用：以JSON格式输出结果
- 用途：
- 程序解析
- 自动化处理
- 数据收集
- 作用：实时JSON流输出
- 特点：
- 每间隔输出一次JSON
- 实时监控
- 流式处理
- 作用：包含更多详细信息的JSON流
- 作用：显示更多详细信息
- 用途：调试、详细分析
- 作用：显示iperf3版本信息
- 作用：启用调试输出
- 级别：1-4，数字越大越详细
- 作用：立即刷新输出缓冲区
- 用途：实时查看输出
- 作用：每行输出添加时间戳
- 用途：精确时间记录
显示测试结果和统计信息。
- 显示内容：iperf3原始输出
- 功能：
- 实时更新
- 彩色显示（如果支持）
- 自动滚动
显示关键指标：
- 带宽：测试期间的平均带宽
- 抖动：UDP测试的延迟变化
- 丢包：UDP测试的丢包率
- 时间：测试持续时间
- 数据量：传输的总数据量
- 包数量：发送/接收的包数
- 重传：TCP重传次数
- 发送者：发送端统计
- 接收者：接收端统计
管理历次测试记录。
- 列：时间、模式、协议、服务器、带宽、抖动、丢包、持续时间、并行流、备注
- 排序：点击列标题排序
- 查看：双击查看详情
- 清空历史：删除所有记录
- 导出记录：保存为CSV文件
- 加载历史：从文件加载历史记录
- 保存结果：保存当前测试结果
目的：了解家庭宽带实际速度
步骤：
-
本地回环测试（验证工具）
- 模式：客户端
- 服务器：127.0.0.1
- 时间：10秒
- 结果应该接近硬件极限（>1Gbps）
-
局域网测试（排除互联网因素）
- 用另一台电脑作为服务器
- 测试有线连接（>900Mbps为正常）
- 测试Wi-Fi连接（根据规格，如Wi-Fi6应>600Mbps）
-
互联网速度测试
- 使用公共iperf3服务器（如 ping.online.net）
- 或自己在外网搭建服务器
- 与运营商宣称速度对比
结果解读：
- 有线局域网：应接近1000Mbps
- Wi-Fi 5：200-500Mbps
- Wi-Fi 6：400-800Mbps
- 家庭宽带：通常是标称的80-90%
目的：找到最佳Wi-Fi位置
方法：
- 服务器放在路由器旁（有线连接）
- 客户端在不同位置测试
- 记录各位置的速度
测试点：
- 同一房间
- 隔一堵墙
- 隔两堵墙
- 楼上/楼下
- 最远角落
目的：测试多人同时使用时的网络情况
方法：
- 设置并行流：4-8
- 模拟场景：一人看4K视频（25Mbps）+一人游戏（10Mbps）+一人下载
- 观察总带宽和每个流的稳定性
新网络部署后验证：
测试项目：
-
单向带宽测试：
- 每个链路单独测试
- 使用标准参数
- 记录基准值
-
双向带宽测试：
- 测试全双工性能
- 验证交换机性能
-
多流并发测试：
- 测试设备并发处理能力
- 模拟多用户场景
-
长时间稳定性测试：
- 持续测试30分钟
- 检查是否有波动
验收标准：
- 达到合同规定带宽的95%以上
- 丢包率<0.1%
- 抖动<5ms（对于实时应用）
网络变慢时排查：
排查步骤：
-
分层测试法：
第一步：本地回环 → 正常？继续，否则本地问题 第二步：同交换机 → 正常？继续，否则交换机问题 第三步：跨交换机 → 正常？继续，否则互联问题 第四步：跨路由器 → 正常？继续，否则路由问题 第五步：互联网 → 分析结果 -
对比测试法：
- 同时测试两条相似链路
- 对比结果找差异
-
时间分段测试：
- 不同时间段测试
- 找出规律（如每天特定时间慢）
为扩容提供依据：
收集数据：
-
当前使用率：
- 业务高峰时测试
- 记录实际使用带宽
-
增长趋势：
- 每月测试一次
- 分析增长曲线
-
应用需求：
- 测试关键应用所需带宽
- 如视频会议、文件同步等
规划建议：
- 当前使用率 > 70%：开始规划扩容
- 当前使用率 > 85%：紧急扩容
- 考虑20-30%的冗余
测试不同云厂商/区域的性能：
测试矩阵：
维度：
1. 云厂商：AWS、Azure、GCP、阿里云、腾讯云等
2. 区域：不同地理区域
3. 实例类型：不同规格的虚拟机
4. 网络类型：普通网络、增强网络
测试方法：
- 同区域测试：同一数据中心内
- 跨区域测试：不同数据中心之间
- 跨云测试：不同云厂商之间
- 到用户测试：从云到实际用户位置
测试不同类型的云网络：
测试项目：
-
普通网络 vs 增强网络：
- 延迟对比
- 带宽对比
- 稳定性对比
-
公网 vs 内网：
- 安全组影响
- 带宽限制差异
- 成本差异
-
不同带宽包测试：
- 按量计费
- 包年包月
- 共享带宽
测试加密隧道性能：
测试重点：
-
加密开销：
- 测试VPN开启前后的差异
- 计算加密带来的性能损失
-
MTU/MSS调整：
- 测试不同MSS值
- 找到最佳值
-
稳定性测试：
- 长时间测试（24小时）
- 检查是否有断线
测试4G/5G网络性能：
特点：
- 波动较大
- 受信号强度影响
- 有数据限制
测试建议：
- 多点测试：不同地理位置
- 多时段测试：早中晚不同时间
- 移动中测试：移动过程中的变化
测试低功耗网络：
测试类型：
-
NB-IoT/LoRa测试：
- 低带宽测试（几Kbps）
- 高延迟测试（几秒）
- 功耗间接测试
-
连接稳定性：
- 测试重连机制
- 测试弱信号处理
正常输出示例：
[ ID] Interval Transfer Bandwidth Retr
[ 5] 0.00-10.00 sec 1.10 GBytes 944 Mbits/sec 0
[ 5] 10.00-20.00 sec 1.15 GBytes 987 Mbits/sec 1
[ 5] 20.00-30.00 sec 1.12 GBytes 962 Mbits/sec 0
[ 5] 30.00-40.00 sec 1.14 GBytes 979 Mbits/sec 0
[ 5] 40.00-50.00 sec 1.13 GBytes 971 Mbits/sec 2
关键指标：
- 带宽：944-987 Mbits/sec（波动约5%，正常）
- 重传：0-2次（很少，网络质量好）
- 稳定性：各时间段结果接近
正常输出示例：
[ ID] Interval Transfer Bandwidth Jitter Lost/Total Datagrams
[ 5] 0.00-10.00 sec 128 MBytes 107 Mbits/sec 0.512 ms 0/10000 (0%)
[ 5] 10.00-20.00 sec 129 MBytes 108 Mbits/sec 0.498 ms 1/10001 (0.01%)
[ 5] 20.00-30.00 sec 127 MBytes 107 Mbits/sec 0.523 ms 0/10000 (0%)
关键指标：
- 带宽：稳定在107-108 Mbits/sec
- 抖动：0.5ms左右（优秀）
- 丢包：0-0.01%（优秀）
可能原因及排查：
-
硬件限制：
- 检查网卡规格（百兆/千兆/万兆）
- 检查路由器/交换机规格
- 检查网线质量（CAT5e以上）
-
软件限制：
- 检查操作系统网络设置
- 检查防火墙/安全软件
- 检查TCP参数（窗口大小等）
-
网络问题：
- 使用ping测试延迟和丢包
- 使用traceroute查看路径
- 检查是否有带宽限制策略
排查步骤：
步骤1：测试本地回环 → 排除本机问题
步骤2：测试同交换机 → 排除交换机问题
步骤3：测试跨设备 → 定位问题设备
步骤4：分段测试 → 找到瓶颈段
可能原因：
-
网络拥塞：
- 其他应用占用带宽
- 网络设备队列拥塞
- 广播风暴等异常
-
无线网络问题：
- 信号干扰
- 信号衰减
- 信道冲突
-
设备性能问题：
- CPU/内存不足
- 磁盘IO瓶颈
- 网卡驱动问题
解决方法：
- 隔离测试：关闭其他应用
- 多次测试：取平均值
- 分段测试：确定波动来源
错误排查：
-
"连接被拒绝"：
- 服务器未运行
- 端口错误
- 防火墙阻挡
-
"连接超时"：
- 网络不通
- 路由问题
- 防火墙丢弃
-
"地址无法解析"：
- 域名错误
- DNS问题
- 本地hosts文件问题
检查清单：
□ 服务器是否运行？
□ 端口是否正确？
□ 防火墙是否允许？
□ 网络是否连通？
□ 地址是否正确？
优化措施：
-
增大TCP窗口：
公式：窗口大小 ≥ 带宽 × RTT 示例：100Mbps带宽，100ms RTT 窗口 ≥ 100Mbps × 0.1s = 10Mbits = 1.25MB -
启用窗口缩放：
- 现代TCP自动支持
- 确保两端都支持
-
调整TCP算法：
- 尝试不同拥塞控制算法
- 如BBR、CUBIC等
优化措施：
-
减小MSS：
- 大包更容易被丢弃
- 尝试536-1024的小包
-
启用SACK：
- 选择性确认
- 提高重传效率
-
调整重传策略：
- 减少快速重传阈值
- 调整RTO参数
优化措施：
-
使用UDP测试：
- TCP对无线不友好
- UDP更能反映真实性能
-
调整包大小：
- 无线有额外开销
- 找到最佳包大小
-
考虑无线特性：
- 信号强度影响
- 干扰避免
- MIMO利用
import subprocess
import json
import time
from datetime import datetime
def run_iperf_test(config):
"""运行iperf测试"""
cmd = ["iperf3", "-c", config["server"], "-t", str(config["duration"])]
if config.get("parallel"):
cmd.extend(["-P", str(config["parallel"])])
if config.get("bandwidth"):
cmd.extend(["-b", config["bandwidth"]])
# 运行测试
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=config["duration"] + 10
)
return parse_iperf_output(result.stdout)
def parse_iperf_output(output):
"""解析iperf输出"""
lines = output.split('\n')
results = {
"bandwidth": 0,
"jitter": 0,
"loss": 0,
"timestamp": datetime.now().isoformat()
}
for line in lines:
if "bits/sec" in line and "sender" in line:
# 解析带宽
if "Gbits/sec" in line:
match = re.search(r'([\d\.]+)\s+Gbits/sec', line)
if match:
results["bandwidth"] = float(match.group(1)) * 1000
elif "Mbits/sec" in line:
# ... 类似处理
# 解析其他指标...
return results
def main():
# 测试配置
tests = [
{"server": "192.168.1.100", "duration": 30, "parallel": 1},
{"server": "192.168.1.100", "duration": 30, "parallel": 4},
{"server": "192.168.1.100", "duration": 30, "parallel": 8},
]
all_results = []
for test_config in tests:
print(f"运行测试: {test_config}")
result = run_iperf_test(test_config)
all_results.append(result)
time.sleep(5) # 测试间隔
# 保存结果
with open("test_results.json", "w") as f:
json.dump(all_results, f, indent=2)
print("测试完成，结果已保存")
if __name__ == "__main__":
main()#!/usr/bin/env python3
"""
网络质量定期监控脚本
每天定时测试并记录
"""
import schedule
import time
from iperf_automation import run_iperf_test
def daily_monitor():
"""每日监控任务"""
print(f"开始每日网络监控 - {time.ctime()}")
# 测试到多个目标
targets = [
{"name": "本地服务器", "server": "192.168.1.1", "port": 5201},
{"name": "互联网测试点1", "server": "ping.online.net", "port": 5201},
{"name": "互联网测试点2", "server": "iperf.he.net", "port": 5201},
]
results = []
for target in targets:
try:
result = run_iperf_test({
"server": target["server"],
"duration": 60,
"parallel": 4
})
result["target"] = target["name"]
results.append(result)
except Exception as e:
print(f"测试{target['name']}失败: {e}")
# 保存结果
save_results(results)
print("每日监控完成")
def save_results(results):
"""保存结果到数据库或文件"""
# 这里可以保存到数据库、文件或发送到监控系统
pass
# 设置定时任务
schedule.every().day.at("02:00").do(daily_monitor) # 每天凌晨2点
schedule.every().hour.do(lambda: run_quick_test()) # 每小时快速测试
def run_quick_test():
"""快速测试"""
pass
if __name__ == "__main__":
print("网络监控服务启动...")
while True:
schedule.run_pending()
time.sleep(60)from prometheus_client import start_http_server, Gauge
import time
# 创建指标
BANDWIDTH_GAUGE = Gauge('network_bandwidth_mbps',
'Network bandwidth in Mbps',
['direction', 'server'])
LATENCY_GAUGE = Gauge('network_latency_ms',
'Network latency in ms',
['server'])
LOSS_GAUGE = Gauge('network_packet_loss_percent',
'Packet loss percentage',
['server'])
def update_metrics(server, results):
"""更新Prometheus指标"""
BANDWIDTH_GAUGE.labels(direction='upload', server=server).set(
results.get('upload_bandwidth', 0)
)
BANDWIDTH_GAUGE.labels(direction='download', server=server).set(
results.get('download_bandwidth', 0)
)
LATENCY_GAUGE.labels(server=server).set(
results.get('latency', 0)
)
LOSS_GAUGE.labels(server=server).set(
results.get('loss', 0)
)
def monitoring_loop():
"""监控循环"""
start_http_server(8000) # Prometheus metrics端口
servers = ["server1", "server2", "server3"]
while True:
for server in servers:
try:
# 运行iperf测试
results = run_iperf_test_to(server)
# 更新指标
update_metrics(server, results)
except Exception as e:
print(f"监控{server}失败: {e}")
time.sleep(300) # 每5分钟一次创建监控仪表板，包含：
- 实时带宽图表
- 历史趋势图
- 网络质量评分
- 告警面板
- 地理分布图
class BatchTester:
def __init__(self, config_file):
self.configs = self.load_configs(config_file)
self.results = []
def load_configs(self, config_file):
"""加载测试配置"""
# 从JSON/YAML文件加载
pass
def run_all_tests(self):
"""运行所有测试"""
for test_name, config in self.configs.items():
print(f"运行测试: {test_name}")
try:
result = self.run_single_test(config)
result["test_name"] = test_name
result["timestamp"] = datetime.now().isoformat()
self.results.append(result)
# 可选：测试间等待
time.sleep(config.get("interval", 10))
except Exception as e:
print(f"测试{test_name}失败: {e}")
self.results.append({
"test_name": test_name,
"error": str(e),
"status": "failed"
})
def run_single_test(self, config):
"""运行单个测试"""
# 具体测试逻辑
pass
def generate_report(self, format="html"):
"""生成报告"""
if format == "html":
return self.generate_html_report()
elif format == "pdf":
return self.generate_pdf_report()
elif format == "markdown":
return self.generate_markdown_report()
def generate_html_report(self):
"""生成HTML报告"""
template = """
<html>
<head>
<title>网络测试报告</title>
<style>
table { border-collapse: collapse; width: 100%; }
th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
th { background-color: #f2f2f2; }
.pass { background-color: #d4edda; }
.fail { background-color: #f8d7da; }
.warning { background-color: #fff3cd; }
</style>
</head>
<body>
<h1>网络测试报告</h1>
<p>生成时间: {{ timestamp }}</p>
<h2>测试概览</h2>
<table>
<tr>
<th>测试名称</th>
<th>带宽(Mbps)</th>
<th>延迟(ms)</th>
<th>丢包率(%)</th>
<th>状态</th>
</tr>
{% for result in results %}
<tr class="{{ result.status }}">
<td>{{ result.test_name }}</td>
<td>{{ result.bandwidth | default('N/A') }}</td>
<td>{{ result.latency | default('N/A') }}</td>
<td>{{ result.loss | default('N/A') }}</td>
<td>{{ result.status }}</td>
</tr>
{% endfor %}
</table>
<h2>详细结果</h2>
{% for result in results %}
<h3>{{ result.test_name }}</h3>
<pre>{{ result.raw_output }}</pre>
{% endfor %}
</body>
</html>
"""
# 使用模板引擎渲染
# ...
return html_content# tests.yaml
tests:
# 基础连通性测试
basic_connectivity:
description: "基础连通性测试"
server: "192.168.1.1"
port: 5201
duration: 30
parallel: 1
expected:
bandwidth_min: 800 # Mbps
latency_max: 10 # ms
loss_max: 0.1 # %
# 压力测试
stress_test:
description: "多流压力测试"
server: "192.168.1.1"
duration: 300
parallel: 16
bandwidth: "1G"
expected:
bandwidth_min: 900
# 双向测试
bidirectional_test:
description: "双向带宽测试"
server: "192.168.1.1"
duration: 60
parallel: 8
bidirectional: true
# 不同协议测试
protocol_tests:
- name: "tcp_test"
protocol: "tcp"
duration: 30
- name: "udp_test"
protocol: "udp"
duration: 30
bandwidth: "100M"
schedule:
daily: "02:00"
weekly: "sunday 03:00"
monthly: "1 04:00"
reporting:
formats: ["html", "pdf", "email"]
recipients: ["admin@example.com"]
thresholds:
warning: 80%
critical: 90%class NetworkQualityScorer:
def __init__(self, weights=None):
# 默认权重
self.weights = weights or {
"bandwidth": 0.4, # 带宽权重 40%
"latency": 0.3, # 延迟权重 30%
"loss": 0.2, # 丢包权重 20%
"jitter": 0.1, # 抖动权重 10%
}
# 基准值（可根据网络类型调整）
self.baselines = {
"lan": {
"bandwidth": 1000, # 1Gbps
"latency": 1, # 1ms
"loss": 0, # 0%
"jitter": 0.1, # 0.1ms
},
"wan": {
"bandwidth": 100, # 100Mbps
"latency": 50, # 50ms
"loss": 0.1, # 0.1%
"jitter": 5, # 5ms
},
"internet": {
"bandwidth": 50, # 50Mbps
"latency": 100, # 100ms
"loss": 0.5, # 0.5%
"jitter": 10, # 10ms
}
}
def calculate_score(self, results, network_type="wan"):
"""计算网络质量分数（0-100）"""
baseline = self.baselines[network_type]
scores = {}
# 带宽得分（越高越好）
bandwidth_score = min(
100,
(results.get("bandwidth", 0) / baseline["bandwidth"]) * 100
)
# 延迟得分（越低越好，需要转换）
latency = results.get("latency", float('inf'))
if latency <= baseline["latency"]:
latency_score = 100
else:
# 延迟加倍，分数减半
latency_score = max(
0,
100 * (baseline["latency"] / latency)
)
# 丢包得分（0丢包得100分）
loss = results.get("loss", 100) # 默认100%丢包
if loss <= baseline["loss"]:
loss_score = 100
else:
loss_score = max(
0,
100 * (1 - loss / 100) # 转换为比例
)
# 抖动得分
jitter = results.get("jitter", float('inf'))
if jitter <= baseline["jitter"]:
jitter_score = 100
else:
jitter_score = max(
0,
100 * (baseline["jitter"] / jitter)
)
# 加权总分
total_score = (
bandwidth_score * self.weights["bandwidth"] +
latency_score * self.weights["latency"] +
loss_score * self.weights["loss"] +
jitter_score * self.weights["jitter"]
)
return {
"total": round(total_score, 1),
"components": {
"bandwidth": round(bandwidth_score, 1),
"latency": round(latency_score, 1),
"loss": round(loss_score, 1),
"jitter": round(jitter_score, 1),
},
"details": {
"bandwidth_actual": results.get("bandwidth"),
"latency_actual": results.get("latency"),
"loss_actual": results.get("loss"),
"jitter_actual": results.get("jitter"),
}
}
def get_quality_level(self, score):
"""根据分数获取质量等级"""
if score >= 90:
return {"level": "优秀", "color": "green", "emoji": "✅"}
elif score >= 80:
return {"level": "良好", "color": "blue", "emoji": "👍"}
elif score >= 60:
return {"level": "一般", "color": "yellow", "emoji": "⚠️"}
elif score >= 40:
return {"level": "较差", "color": "orange", "emoji": "🔶"}
else:
return {"level": "差", "color": "red", "emoji": "❌"}
def generate_quality_report(self, test_results):
"""生成质量报告"""
report = {
"timestamp": datetime.now().isoformat(),
"tests": [],
"summary": {
"average_score": 0,
"best_test": None,
"worst_test": None,
"recommendations": []
}
}
total_score = 0
best_score = -1
worst_score = 101
for test in test_results:
score_result = self.calculate_score(test["metrics"])
quality = self.get_quality_level(score_result["total"])
test_report = {
"name": test["name"],
"score": score_result["total"],
"quality": quality,
"details": score_result
}
report["tests"].append(test_report)
# 更新统计
total_score += score_result["total"]
if score_result["total"] > best_score:
best_score = score_result["total"]
report["summary"]["best_test"] = test["name"]
if score_result["total"] < worst_score:
worst_score = score_result["total"]
report["summary"]["worst_test"] = test["name"]
# 计算平均分
if report["tests"]:
report["summary"]["average_score"] = total_score / len(report["tests"])
# 生成建议
report["summary"]["recommendations"] = self.generate_recommendations(report)
return report
def generate_recommendations(self, report):
"""根据测试结果生成改进建议"""
recommendations = []
for test in report["tests"]:
details = test["details"]["details"]
components = test["details"]["components"]
# 带宽相关建议
if components["bandwidth"] < 80:
actual = details.get("bandwidth_actual", 0)
expected = self.baselines["wan"]["bandwidth"]
recommendations.append(
f"测试 '{test['name']}' 带宽较低 ({actual}Mbps vs 预期{expected}Mbps)。"
"建议检查：1.网线质量 2.网络设备规格 3.是否有带宽限制"
)
# 延迟相关建议
if components["latency"] < 80:
actual = details.get("latency_actual", 0)
recommendations.append(
f"测试 '{test['name']}' 延迟较高 ({actual}ms)。"
"建议：1.优化路由 2.减少网络跳数 3.检查网络设备负载"
)
# 丢包相关建议
if components["loss"] < 80:
actual = details.get("loss_actual", 0)
recommendations.append(
f"测试 '{test['name']}' 丢包率较高 ({actual}%)。"
"建议：1.检查网络线路 2.优化MTU设置 3.检查网络设备错误计数"
)
# 去重
return list(set(recommendations))def create_quality_dashboard(report):
"""创建质量仪表板"""
import matplotlib.pyplot as plt
import pandas as pd
# 准备数据
tests = [t["name"] for t in report["tests"]]
scores = [t["score"] for t in report["tests"]]
colors = [t["quality"]["color"] for t in report["tests"]]
# 创建图表
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# 1. 总分柱状图
axes[0, 0].bar(tests, scores, color=colors)
axes[0, 0].set_title("各测试总分")
axes[0, 0].set_ylabel("分数 (0-100)")
axes[0, 0].set_ylim(0, 100)
axes[0, 0].tick_params(axis='x', rotation=45)
# 2. 各指标雷达图
categories = ['带宽', '延迟', '丢包', '抖动']
N = len(categories)
for i, test in enumerate(report["tests"]):
values = [
test["details"]["components"]["bandwidth"],
test["details"]["components"]["latency"],
test["details"]["components"]["loss"],
test["details"]["components"]["jitter"]
]
values += values[:1] # 闭合雷达图
angles = [n / float(N) * 2 * 3.14159 for n in range(N)]
angles += angles[:1]
ax = axes[0, 1]
ax.plot(angles, values, label=test["name"])
ax.fill(angles, values, alpha=0.1)
axes[0, 1].set_title("各测试指标雷达图")
axes[0, 1].legend()
# 3. 时间趋势图（如果有历史数据）
# ...
# 4. 质量分布饼图
quality_counts = {}
for test in report["tests"]:
level = test["quality"]["level"]
quality_counts[level] = quality_counts.get(level, 0) + 1
axes[1, 0].pie(
quality_counts.values(),
labels=quality_counts.keys(),
autopct='%1.1f%%',
colors=['green', 'blue', 'yellow', 'orange', 'red'][:len(quality_counts)]
)
axes[1, 0].set_title("质量等级分布")
# 5. 详细数据表格
axes[1, 1].axis('tight')
axes[1, 1].axis('off')
table_data = []
for test in report["tests"]:
table_data.append([
test["name"],
f"{test['score']:.1f}",
test["quality"]["level"],
f"{test['details']['details'].get('bandwidth_actual', 'N/A')}",
f"{test['details']['details'].get('latency_actual', 'N/A')}ms",
f"{test['details']['details'].get('loss_actual', 'N/A')}%"
])
table = axes[1, 1].table(
cellText=table_data,
colLabels=["测试", "分数", "等级", "带宽", "延迟", "丢包"],
loc='center',
cellLoc='center'
)
table.auto_set_font_size(False)
table.set_fontsize(9)
table.scale(1, 1.5)
plt.tight_layout()
plt.savefig("network_quality_report.png", dpi=300, bbox_inches='tight')
plt.show()
return fig不要将iperf服务器长期暴露在公网，除非：
- 有认证机制
- 有访问控制
- 有日志监控
- 有带宽限制
推荐的安全配置：
# 使用认证
iperf3 -s --username admin --rsa-private-key-path /path/to/key
# 限制带宽
iperf3 -s --server-bitrate-limit 100M
# 限制连接时间
iperf3 -s --idle-timeout 300 --server-max-duration 3600
# 使用非标准端口
iperf3 -s -p 15201
# 绑定特定IP
iperf3 -s -B 192.168.1.100必要的最小规则：
# 只允许特定IP访问
iptables -A INPUT -p tcp --dport 5201 -s 192.168.1.0/24 -j ACCEPT
iptables -A INPUT -p tcp --dport 5201 -j DROP
# 限制连接速率
iptables -A INPUT -p tcp --dport 5201 -m limit --limit 10/min -j ACCEPT
iptables -A INPUT -p tcp --dport 5201 -j DROP
测试前检查清单：
□ 1. 确认网络拓扑
□ 2. 记录设备型号和固件版本
□ 3. 检查线缆质量（CAT5e以上）
□ 4. 确认端口协商状态（千兆/万兆）
□ 5. 关闭节能模式（EEE）
□ 6. 更新网卡驱动
□ 7. 调整TCP参数（如果需要）
□ 8. 关闭无关应用程序
□ 9. 清理ARP缓存
□ 10. 记录测试前计数器
科学的测试步骤：
-
基线测试：
- 最低负载测试
- 单流测试
- 记录基准值
-
增量测试：
- 逐步增加负载
- 观察性能变化
- 找到拐点
-
压力测试：
- 长时间满负载
- 测试稳定性
- 观察错误率
-
恢复测试：
- 停止压力后
- 测试恢复时间
- 验证弹性
确保结果可信的方法：
- 重复测试：至少3次，取平均值
- 交叉验证：用不同工具验证
- 双向验证：测试两个方向
- 分段验证：分段测试定位问题
- 时间验证：不同时间测试
避免这些常见错误：
-
忽视CPU性能：
- iperf3是单线程的
- 高速测试需要多核CPU
- 使用
-A参数绑定CPU
-
忘记窗口缩放：
- 高速高延迟需要大窗口
- 确保窗口缩放启用
- 调整系统参数
-
MTU不匹配：
- 检查路径MTU
- 避免分片
- 适当调整MSS
-
缓冲区不足：
- 增加socket缓冲区
- 调整系统参数
- 验证实际使用
测试时的注意事项：
-
不要在生产高峰测试：
- 选择非业务时间
- 提前通知相关人员
- 做好回滚准备
-
注意带宽占用：
- 从低带宽开始
- 逐步增加
- 监控影响
-
保存原始数据：
- 保存iperf原始输出
- 记录测试环境
- 保存配置参数
-
考虑测试影响：
- 对网络设备的影响
- 对其他应用的影响
- 对监控系统的影响
错误：connect failed: Connection refused
可能原因：
- 服务器未运行
- 端口错误
- 防火墙阻挡
解决步骤：
1. 检查服务器是否运行：ps aux | grep iperf3
2. 检查端口是否正确：netstat -tlnp | grep 5201
3. 检查防火墙：iptables -L -n
4. 尝试telnet测试：telnet 服务器IP 端口
错误：connect failed: Connection timed out
可能原因：
- 网络不通
- 路由问题
- 中间设备丢弃
解决步骤：
1. ping测试连通性
2. traceroute查看路径
3. 检查中间设备ACL
4. 检查MTU问题
问题：速度远低于预期
排查步骤：
步骤1：测试本地回环 → 验证工具
步骤2：测试同交换机 → 验证本地网络
步骤3：逐跳测试 → 定位问题段
步骤4：检查设备计数器 → 查找错误
常见原因：
-
网卡协商问题：
# 检查协商状态 ethtool eth0 # 强制千兆 ethtool -s eth0 speed 1000 duplex full autoneg off
-
TCP参数问题：
# 检查窗口大小 sysctl net.ipv4.tcp_window_scaling # 检查缓冲区 sysctl net.core.rmem_max sysctl net.core.wmem_max
-
系统限制问题：
# 检查文件描述符限制 ulimit -n # 检查CPU频率 cpupower frequency-info
问题：测试结果波动大
可能原因：
- 网络拥塞
- 无线干扰
- 设备负载
- 其他应用影响
诊断方法：
-
时间分析：
- 不同时间测试
- 寻找规律
-
对比分析：
- 与其他链路对比
- 与历史数据对比
-
监控分析：
- 实时监控网络流量
- 监控设备负载
捕获iperf流量分析：
# 在服务器端捕获
tcpdump -i eth0 -w iperf.pcap port 5201
# 分析TCP序列号
tcpdump -r iperf.pcap -n 'tcp port 5201' | head -20
# 查看重传
tcpdump -r iperf.pcap -n 'tcp port 5201 and tcp[tcpflags] & (tcp-syn|tcp-ack) == tcp-ack'综合性能分析：
# 实时监控
nethogs eth0 # 查看每个连接带宽
iftop -i eth0 # 查看实时流量
iotop # 查看磁盘IO
htop # 查看CPU/内存
# 网络统计
sar -n DEV 1 # 网络设备统计
sar -n TCP 1 # TCP统计
sar -n EDEV 1 # 错误统计深入分析建议：
- Wireshark：详细协议分析
- perf：系统性能分析
- systemtap：内核级分析
- bpftrace：现代性能分析
立即停止测试：
# 快速停止所有iperf进程
pkill -9 iperf3
# 清理连接
ss -K dst 服务器IP dport = 5201恢复网络：
# 刷新ARP
ip neigh flush dev eth0
# 清理conntrack
conntrack -F
# 重启网络服务
systemctl restart network故障时收集信息：
# 收集系统信息
dmesg -T > dmesg.log
journalctl -xe > journal.log
# 收集网络信息
ip addr > ip_addr.log
ip route > ip_route.log
ss -tulnp > ss.log
netstat -s > netstat_s.log
# 收集设备信息
ethtool -S eth0 > ethtool_S.log
ethtool -k eth0 > ethtool_k.log
# 打包所有日志
tar czf debug_$(date +%Y%m%d_%H%M%S).tar.gz *.log# plugin_base.py
import abc
from typing import Dict, Any
class IperfPlugin(abc.ABC):
"""插件基类"""
def __init__(self, name, version):
self.name = name
self.version = version
@abc.abstractmethod
def before_test(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""测试前调用，可以修改配置"""
pass
@abc.abstractmethod
def during_test(self, output_line: str) -> None:
"""测试过程中每行输出调用"""
pass
@abc.abstractmethod
def after_test(self, results: Dict[str, Any]) -> Dict[str, Any]:
"""测试后调用，可以处理结果"""
pass
@abc.abstractmethod
def get_ui_widget(self):
"""返回插件的UI部件"""
pass
# plugin_manager.py
class PluginManager:
def __init__(self):
self.plugins = {}
self.load_plugins()
def load_plugins(self):
"""动态加载插件"""
import importlib
import pkgutil
plugin_packages = ['plugins']
for package_name in plugin_packages:
try:
package = importlib.import_module(package_name)
for _, module_name, _ in pkgutil.iter_modules(package.__path__):
module = importlib.import_module(f"{package_name}.{module_name}")
for attr_name in dir(module):
attr = getattr(module, attr_name)
if (isinstance(attr, type) and
issubclass(attr, IperfPlugin) and
attr != IperfPlugin):
plugin_instance = attr()
self.plugins[plugin_instance.name] = plugin_instance
except ImportError:
continue
def run_before_hooks(self, config):
"""运行所有before_test钩子"""
for plugin in self.plugins.values():
config = plugin.before_test(config)
return config
def run_during_hooks(self, output_line):
"""运行所有during_test钩子"""
for plugin in self.plugins.values():
plugin.during_test(output_line)
def run_after_hooks(self, results):
"""运行所有after_test钩子"""
for plugin in self.plugins.values():
results = plugin.after_test(results)
return results# plugins/bandwidth_monitor.py
import time
from typing import Dict, Any
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QLabel, QProgressBar
class BandwidthMonitorPlugin(IperfPlugin):
"""带宽监控插件"""
def __init__(self):
super().__init__("带宽监控", "1.0")
self.start_time = None
self.total_bytes = 0
self.max_bandwidth = 0
self.ui_widget = None
def before_test(self, config: Dict[str, Any]) -> Dict[str, Any]:
self.start_time = time.time()
self.total_bytes = 0
self.max_bandwidth = 0
return config
def during_test(self, output_line: str) -> None:
# 解析输出行，提取带宽信息
if "Mbits/sec" in output_line or "Gbits/sec" in output_line:
# 这里添加解析逻辑
pass
def after_test(self, results: Dict[str, Any]) -> Dict[str, Any]:
if self.start_time:
duration = time.time() - self.start_time
avg_bandwidth = (self.total_bytes * 8) / duration / 1e6 # Mbps
results["plugin_stats"] = {
"total_bytes": self.total_bytes,
"average_bandwidth_mbps": avg_bandwidth,
"max_bandwidth_mbps": self.max_bandwidth,
"duration": duration
}
return results
def get_ui_widget(self):
if not self.ui_widget:
self.ui_widget = QWidget()
layout = QVBoxLayout()
self.current_label = QLabel("当前带宽: --")
self.average_label = QLabel("平均带宽: --")
self.max_label = QLabel("最大带宽: --")
self.progress_bar = QProgressBar()
layout.addWidget(self.current_label)
layout.addWidget(self.average_label)
layout.addWidget(self.max_label)
layout.addWidget(self.progress_bar)
self.ui_widget.setLayout(layout)
return self.ui_widget# api_server.py
from flask import Flask, request, jsonify
import threading
import time
app = Flask(__name__)
class APIServer:
def __init__(self, iperf_gui):
self.gui = iperf_gui
self.test_thread = None
self.current_test = None
def run(self, host='0.0.0.0', port=5000):
"""启动API服务器"""
app.run(host=host, port=port, threaded=True)
@app.route('/api/test/start', methods=['POST'])
def start_test():
"""启动测试"""
data = request.json
if self.current_test and self.current_test.is_alive():
return jsonify({"error": "测试正在进行中"}), 409
# 准备测试参数
params = self.prepare_params(data)
# 在新线程中运行测试
self.test_thread = threading.Thread(
target=self.run_test_in_thread,
args=(params,)
)
self.test_thread.start()
return jsonify({
"status": "started",
"test_id": str(int(time.time()))
})
@app.route('/api/test/stop', methods=['POST'])
def stop_test():
"""停止测试"""
if self.current_test:
self.current_test.stop()
return jsonify({"status": "stopped"})
else:
return jsonify({"error": "没有正在运行的测试"}), 404
@app.route('/api/test/status', methods=['GET'])
def get_status():
"""获取测试状态"""
if self.current_test and self.current_test.is_alive():
return jsonify({
"status": "running",
"progress": self.current_test.get_progress(),
"results": self.current_test.get_partial_results()
})
else:
return jsonify({"status": "stopped"})
@app.route('/api/config', methods=['GET', 'POST'])
def config():
"""配置管理"""
if request.method == 'GET':
return jsonify(self.gui.get_current_config())
else:
config = request.json
self.gui.apply_config(config)
return jsonify({"status": "applied"})
def run_test_in_thread(self, params):
"""在线程中运行测试"""
self.current_test = self.gui.start_test_with_params(params)
while self.current_test.is_alive():
time.sleep(0.1)
self.current_test = None# websocket_server.py
from flask_socketio import SocketIO, emit
import json
socketio = SocketIO(app, cors_allowed_origins="*")
@socketio.on('connect')
def handle_connect():
print('客户端连接')
emit('connected', {'data': '连接成功'})
@socketio.on('start_test')
def handle_start_test(data):
"""通过WebSocket启动测试"""
test_id = start_test_async(data)
emit('test_started', {'test_id': test_id})
# 实时推送结果
def send_updates():
while test_is_running(test_id):
results = get_latest_results(test_id)
emit('test_update', results)
socketio.sleep(0.5)
socketio.start_background_task(send_updates)
@socketio.on('stop_test')
def handle_stop_test(data):
test_id = data.get('test_id')
stop_test(test_id)
emit('test_stopped', {'test_id': test_id})# report_generator.py
from jinja2 import Environment, FileSystemLoader
import pdfkit
import pandas as pd
class ReportGenerator:
def __init__(self, template_dir='templates'):
self.env = Environment(
loader=FileSystemLoader(template_dir),
extensions=['jinja2.ext.do']
)
# 注册自定义过滤器
self.env.filters['format_bandwidth'] = self.format_bandwidth
self.env.filters['format_time'] = self.format_time
self.env.filters['quality_color'] = self.quality_color
def generate_report(self, data, template_name, output_format='html'):
"""生成报告"""
template = self.env.get_template(template_name)
# 渲染模板
html_content = template.render(**data)
if output_format == 'html':
return html_content
elif output_format == 'pdf':
return self.html_to_pdf(html_content)
elif output_format == 'docx':
return self.html_to_docx(html_content)
else:
raise ValueError(f"不支持的格式: {output_format}")
def format_bandwidth(self, value, unit='auto'):
"""格式化带宽显示"""
if unit == 'auto':
if value >= 1000:
return f"{value/1000:.2f} Gbps"
elif value >= 1:
return f"{value:.2f} Mbps"
else:
return f"{value*1000:.2f} Kbps"
elif unit == 'G':
return f"{value/1000:.2f} Gbps"
elif unit == 'M':
return f"{value:.2f} Mbps"
elif unit == 'K':
return f"{value*1000:.2f} Kbps"
def format_time(self, seconds):
"""格式化时间显示"""
if seconds < 60:
return f"{seconds:.1f} 秒"
elif seconds < 3600:
minutes = int(seconds // 60)
secs = seconds % 60
return f"{minutes} 分 {secs:.0f} 秒"
else:
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
return f"{hours} 小时 {minutes} 分"
def quality_color(self, score):
"""根据分数返回颜色"""
if score >= 90:
return "#28a745" # 绿色
elif score >= 80:
return "#17a2b8" # 蓝色
elif score >= 60:
return "#ffc107" # 黄色
elif score >= 40:
return "#fd7e14" # 橙色
else:
return "#dc3545" # 红色
def html_to_pdf(self, html_content):
"""HTML转PDF"""
options = {
'page-size': 'A4',
'margin-top': '0.75in',
'margin-right': '0.75in',
'margin-bottom': '0.75in',
'margin-left': '0.75in',
'encoding': "UTF-8",
'no-outline': None,
'enable-local-file-access': None
}
pdf = pdfkit.from_string(html_content, False, options=options)
return pdf
def generate_comparison_report(self, results_list, baseline=None):
"""生成对比报告"""
# 创建DataFrame
df = pd.DataFrame(results_list)
# 计算统计
stats = {
'测试次数': len(df),
'平均带宽': df['bandwidth'].mean(),
'最大带宽': df['bandwidth'].max(),
'最小带宽': df['bandwidth'].min(),
'带宽标准差': df['bandwidth'].std(),
'平均延迟': df['latency'].mean(),
'平均丢包': df['loss'].mean(),
}
if baseline:
stats['与基线差异'] = {
'带宽': ((df['bandwidth'].mean() - baseline['bandwidth']) / baseline['bandwidth']) * 100,
'延迟': ((df['latency'].mean() - baseline['latency']) / baseline['latency']) * 100,
}
# 生成报告数据
report_data = {
'statistics': stats,
'results': df.to_dict('records'),
'charts': {
'bandwidth_trend': self.create_bandwidth_chart(df),
'quality_distribution': self.create_quality_distribution(df),
'comparison_chart': self.create_comparison_chart(df, baseline) if baseline else None
}
}
return self.generate_report(report_data, 'comparison_report.html')
def create_bandwidth_chart(self, df):
"""创建带宽趋势图数据"""
# 这里生成图表数据，可以是Plotly、Chart.js等格式
chart_data = {
'labels': df['timestamp'].tolist(),
'datasets': [{
'label': '带宽 (Mbps)',
'data': df['bandwidth'].tolist(),
'borderColor': 'rgb(75, 192, 192)',
'backgroundColor': 'rgba(75, 192, 192, 0.2)',
}]
}
return chart_data<!-- templates/comparison_report.html -->
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>网络测试对比报告</title>
<script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
<style>
body {
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
line-height: 1.6;
color: #333;
max-width: 1200px;
margin: 0 auto;
padding: 20px;
}
.header {
text-align: center;
border-bottom: 2px solid #007bff;
padding-bottom: 20px;
margin-bottom: 30px;
}
.summary-cards {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(250px, 1fr));
gap: 20px;
margin-bottom: 30px;
}
.card {
background: #f8f9fa;
border-radius: 8px;
padding: 20px;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
.card h3 {
color: #495057;
margin-top: 0;
}
.stat-value {
font-size: 2em;
font-weight: bold;
color: #007bff;
}
.stat-label {
color: #6c757d;
font-size: 0.9em;
}
table {
width: 100%;
border-collapse: collapse;
margin-bottom: 30px;
}
th, td {
border: 1px solid #dee2e6;
padding: 12px;
text-align: left;
}
th {
background-color: #007bff;
color: white;
}
tr:nth-child(even) {
background-color: #f8f9fa;
}
.quality-indicator {
display: inline-block;
width: 12px;
height: 12px;
border-radius: 50%;
margin-right: 8px;
}
.chart-container {
margin-bottom: 30px;
height: 400px;
}
.footer {
text-align: center;
margin-top: 50px;
padding-top: 20px;
border-top: 1px solid #dee2e6;
color: #6c757d;
font-size: 0.9em;
}
.recommendations {
background: #e7f5ff;
border-left: 4px solid #007bff;
padding: 15px;
margin: 20px 0;
}
.recommendations h4 {
margin-top: 0;
color: #0056b3;
}
</style>
</head>
<body>
<div class="header">
<h1>网络性能测试对比报告</h1>
<p>生成时间: {{ timestamp }}</p>
<p>测试次数: {{ statistics.测试次数 }} 次</p>
</div>
<div class="summary-cards">
<div class="card">
<h3>带宽性能</h3>
<div class="stat-value">{{ statistics.平均带宽|format_bandwidth }}</div>
<div class="stat-label">平均带宽</div>
<p>范围: {{ statistics.最小带宽|format_bandwidth }} - {{ statistics.最大带宽|format_bandwidth }}</p>
</div>
<div class="card">
<h3>网络质量</h3>
<div class="stat-value">{{ statistics.平均延迟|round(2) }} ms</div>
<div class="stat-label">平均延迟</div>
<p>丢包率: {{ statistics.平均丢包|round(3) }}%</p>
</div>
{% if statistics.与基线差异 %}
<div class="card">
<h3>与基线对比</h3>
<div class="stat-value" style="color: {% if statistics.与基线差异.带宽 > 0 %}#28a745{% else %}#dc3545{% endif %};">
{{ statistics.与基线差异.带宽|round(2) }}%
</div>
<div class="stat-label">带宽变化</div>
<p>延迟变化: {{ statistics.与基线差异.延迟|round(2) }}%</p>
</div>
{% endif %}
</div>
<div class="chart-container">
<canvas id="bandwidthChart"></canvas>
</div>
<h2>详细测试结果</h2>
<table>
<thead>
<tr>
<th>时间</th>
<th>带宽</th>
<th>延迟</th>
<th>丢包</th>
<th>质量评分</th>
<th>备注</th>
</tr>
</thead>
<tbody>
{% for result in results %}
<tr>
<td>{{ result.timestamp }}</td>
<td>{{ result.bandwidth|format_bandwidth }}</td>
<td>{{ result.latency|round(2) }} ms</td>
<td>{{ result.loss|round(3) }}%</td>
<td>
<span class="quality-indicator" style="background-color: {{ result.quality_score|quality_color }};"></span>
{{ result.quality_score|round(1) }}
</td>
<td>{{ result.notes or '' }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% if recommendations %}
<div class="recommendations">
<h4>改进建议</h4>
<ul>
{% for rec in recommendations %}
<li>{{ rec }}</li>
{% endfor %}
</ul>
</div>
{% endif %}
<div class="footer">
<p>报告生成工具: iperf3 图形化工具 v{{ version }}</p>
<p>© {{ year }} 网络测试团队 - 本报告仅用于内部参考</p>
</div>
<script>
// 图表配置
const ctx = document.getElementById('bandwidthChart').getContext('2d');
const chartData = {{ charts.bandwidth_trend|tojson }};
new Chart(ctx, {
type: 'line',
data: chartData,
options: {
responsive: true,
maintainAspectRatio: false,
scales: {
y: {
beginAtZero: true,
title: {
display: true,
text: '带宽 (Mbps)'
}
},
x: {
title: {
display: true,
text: '测试时间'
}
}
}
}
});
</script>
</body>
</html># test_scenarios.py
class TestScenarioLibrary:
"""测试场景库"""
SCENARIOS = {
# 家庭网络场景
"home_network_basic": {
"name": "家庭网络基础测试",
"description": "测试家庭网络基本性能",
"tests": [
{
"name": "局域网有线测试",
"server": "192.168.1.1",
"duration": 30,
"parallel": 1,
"expected_bandwidth": 900 # Mbps
},
{
"name": "Wi-Fi近距离测试",
"server": "192.168.1.1",
"duration": 30,
"parallel": 4,
"wireless": True,
"expected_bandwidth": 300 # Mbps，根据Wi-Fi规格调整
},
{
"name": "互联网下载测试",
"server": "ping.online.net",
"duration": 60,
"parallel": 8,
"expected_bandwidth": "80% of ISP speed"
}
]
},
# 企业网络场景
"enterprise_network_validation": {
"name": "企业网络验收测试",
"description": "企业网络部署后的全面验证",
"tests": [
{
"name": "核心交换机性能测试",
"duration": 300,
"parallel": 32,
"bidirectional": True,
"expected_bandwidth": "line rate"
},
{
"name": "跨部门通信测试",
"duration": 180,
"parallel": 16,
"multiple_servers": True,
"expected_loss": 0
},
{
"name": "互联网出口测试",
"duration": 600,
"parallel": 8,
"long_running": True,
"monitor_stability": True
}
]
},
# 云服务场景
"cloud_migration_validation": {
"name": "云迁移网络验证",
"description": "验证本地到云的网络性能",
"tests": [
{
"name": "本地到云单向带宽",
"direction": "upload",
"duration": 300,
"parallel": 16,
"monitor": ["bandwidth", "loss", "latency"]
},
{
"name": "云到本地单向带宽",
"direction": "download",
"duration": 300,
"parallel": 16
},
{
"name": "双向并发测试",
"duration": 600,
"parallel": 8,
"bidirectional": True,
"stress_test": True
},
{
"name": "不同时段稳定性",
"schedule": ["00:00", "06:00", "12:00", "18:00"],
"duration": 1800,
"monitor_variation": True
}
]
},
# 实时应用场景
"realtime_application_simulation": {
"name": "实时应用模拟测试",
"description": "模拟VoIP、视频会议等实时应用",
"tests": [
{
"name": "VoIP质量测试",
"protocol": "udp",
"bandwidth": "0.1M", # 100Kbps，典型VoIP
"duration": 300,
"monitor": ["jitter", "loss", "latency"],
"expected_jitter": 20, # ms
"expected_loss": 1, # %
"expected_latency": 150 # ms
},
{
"name": "视频会议测试",
"protocol": "udp",
"bandwidth": "4M", # 720p视频
"duration": 600,
"parallel": 4, # 模拟多参会者
"expected_loss": 0.5
},
{
"name": "在线游戏测试",
"protocol": "udp",
"bandwidth": "0.5M",
"duration": 1800,
"packet_size": 512, # 典型游戏包大小
"expected_latency": 50
}
]
},
# 存储网络场景
"storage_network_performance": {
"name": "存储网络性能测试",
"description": "测试NAS、SAN等存储网络性能",
"tests": [
{
"name": "顺序读写测试",
"protocol": "tcp",
"duration": 300,
"window_size": "1M",
"parallel": 1,
"test_pattern": "sequential"
},
{
"name": "随机读写测试",
"protocol": "tcp",
"duration": 300,
"parallel": 32,
"test_pattern": "random",
"io_size": "4K" # 模拟数据库IO
},
{
"name": "混合读写测试",
"protocol": "tcp",
"duration": 600,
"parallel": 16,
"bidirectional": True,
"test_pattern": "mixed"
}
]
},
# 网络安全场景
"network_security_validation": {
"name": "网络安全策略验证",
"description": "验证防火墙、IPS等安全设备的影响",
"tests": [
{
"name": "基础策略测试",
"protocol": "tcp",
"port": 80,
"duration": 60,
"baseline": True
},
{
"name": "IPS深度检测影响",
"protocol": "tcp",
"port": 80,
"duration": 60,
"ips_enabled": True,
"compare_with": "基础策略测试"
},
{
"name": "不同包大小测试",
"protocol": "tcp",
"duration": 60,
"packet_sizes": [64, 512, 1500, 9000],
"compare_performance": True
},
{
"name": "并发连接数测试",
"protocol": "tcp",
"duration": 300,
"parallel": [1, 10, 100, 1000],
"test_connection_limit": True
}
]
}
}
@classmethod
def get_scenario(cls, scenario_id):
"""获取测试场景"""
return cls.SCENARIOS.get(scenario_id)
@classmethod
def list_scenarios(cls):
"""列出所有测试场景"""
return [
{
"id": scenario_id,
"name": config["name"],
"description": config["description"],
"test_count": len(config.get("tests", []))
}
for scenario_id, config in cls.SCENARIOS.items()
]
@classmethod
def execute_scenario(cls, scenario_id, custom_params=None):
"""执行测试场景"""
scenario = cls.get_scenario(scenario_id)
if not scenario:
raise ValueError(f"场景不存在: {scenario_id}")
results = []
all_passed = True
for test_config in scenario["tests"]:
# 合并自定义参数
if custom_params:
test_config.update(custom_params)
print(f"执行测试: {test_config['name']}")
try:
# 执行单个测试
result = cls.execute_single_test(test_config)
# 验证期望值
if "expected_bandwidth" in test_config:
expected = test_config["expected_bandwidth"]
actual = result.get("bandwidth", 0)
if isinstance(expected, str) and "%" in expected:
# 百分比期望，如"80% of ISP speed"
pass # 需要额外逻辑
elif actual < expected * 0.9: # 允许10%误差
result["status"] = "failed"
result["message"] = f"带宽未达预期: {actual} < {expected}"
all_passed = False
else:
result["status"] = "passed"
results.append(result)
except Exception as e:
results.append({
"name": test_config.get("name", "未知测试"),
"status": "error",
"error": str(e)
})
all_passed = False
return {
"scenario_id": scenario_id,
"scenario_name": scenario["name"],
"all_passed": all_passed,
"results": results,
"summary": cls.generate_scenario_summary(results)
}
@classmethod
def execute_single_test(cls, test_config):
"""执行单个测试"""
# 这里调用iperf3执行测试
# 简化示例，实际需要调用iperf3
return {
"name": test_config["name"],
"bandwidth": 950, # 模拟结果
"latency": 12,
"loss": 0,
"duration": test_config.get("duration", 30)
}
@classmethod
def generate_scenario_summary(cls, results):
"""生成场景测试摘要"""
if not results:
return {}
total = len(results)
passed = sum(1 for r in results if r.get("status") == "passed")
failed = sum(1 for r in results if r.get("status") == "failed")
errors = sum(1 for r in results if r.get("status") == "error")
bandwidths = [r.get("bandwidth", 0) for r in results if r.get("bandwidth")]
latencies = [r.get("latency", 0) for r in results if r.get("latency")]
return {
"total_tests": total,
"passed": passed,
"failed": failed,
"errors": errors,
"pass_rate": (passed / total * 100) if total > 0 else 0,
"avg_bandwidth": sum(bandwidths) / len(bandwidths) if bandwidths else 0,
"avg_latency": sum(latencies) / len(latencies) if latencies else 0,
"min_bandwidth": min(bandwidths) if bandwidths else 0,
"max_bandwidth": max(bandwidths) if bandwidths else 0
}# scenario_executor.py
import time
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
class ScenarioExecutor:
"""场景执行器"""
def __init__(self, max_workers=3):
self.max_workers = max_workers
self.executor = ThreadPoolExecutor(max_workers=max_workers)
self.running_tests = {}
self.results = {}
def execute_scenario_async(self, scenario_id, params=None, callback=None):
"""异步执行测试场景"""
future = self.executor.submit(
self._execute_scenario_sync,
scenario_id, params
)
test_id = f"test_{int(time.time())}_{scenario_id}"
self.running_tests[test_id] = {
"future": future,
"scenario_id": scenario_id,
"start_time": datetime.now(),
"callback": callback
}
# 添加完成回调
future.add_done_callback(
lambda f: self._on_scenario_complete(test_id, f)
)
return test_id
def _execute_scenario_sync(self, scenario_id, params):
"""同步执行测试场景"""
from test_scenarios import TestScenarioLibrary
return TestScenarioLibrary.execute_scenario(scenario_id, params)
def _on_scenario_complete(self, test_id, future):
"""场景完成回调"""
if test_id in self.running_tests:
test_info = self.running_tests.pop(test_id)
try:
result = future.result()
result["test_id"] = test_id
result["end_time"] = datetime.now()
self.results[test_id] = result
# 调用用户回调
if test_info["callback"]:
test_info["callback"](result)
# 触发事件
self._emit_event("scenario_completed", result)
except Exception as e:
error_result = {
"test_id": test_id,
"status": "error",
"error": str(e),
"end_time": datetime.now()
}
self.results[test_id] = error_result
self._emit_event("scenario_failed", error_result)
def execute_scenario_sequence(self, scenario_ids, params_list=None,
interval=30, stop_on_error=False):
"""按顺序执行多个场景"""
if params_list is None:
params_list = [{}] * len(scenario_ids)
sequence_id = f"seq_{int(time.time())}"
sequence_results = {
"sequence_id": sequence_id,
"start_time": datetime.now(),
"scenarios": [],
"status": "running"
}
all_passed = True
for i, (scenario_id, params) in enumerate(zip(scenario_ids, params_list)):
print(f"执行场景 {i+1}/{len(scenario_ids)}: {scenario_id}")
try:
result = self._execute_scenario_sync(scenario_id, params)
result["order"] = i + 1
sequence_results["scenarios"].append(result)
if not result.get("all_passed", True):
all_passed = False
if stop_on_error:
print(f"场景 {scenario_id} 失败，停止序列执行")
break
# 场景间间隔
if i < len(scenario_ids) - 1:
print(f"等待 {interval} 秒后执行下一个场景...")
time.sleep(interval)
except Exception as e:
error_result = {
"scenario_id": scenario_id,
"order": i + 1,
"status": "error",
"error": str(e)
}
sequence_results["scenarios"].append(error_result)
all_passed = False
if stop_on_error:
break
sequence_results["end_time"] = datetime.now()
sequence_results["status"] = "completed" if all_passed else "failed"
sequence_results["all_passed"] = all_passed
# 保存序列结果
self.results[sequence_id] = sequence_results
self._emit_event("sequence_completed", sequence_results)
return sequence_results
def get_test_status(self, test_id):
"""获取测试状态"""
if test_id in self.running_tests:
future = self.running_tests[test_id]["future"]
if future.running():
return {"status": "running"}
elif future.done():
return {"status": "done"}
else:
return {"status": "pending"}
elif test_id in self.results:
result = self.results[test_id]
return {
"status": "completed",
"result": result
}
else:
return {"status": "not_found"}
def get_all_results(self):
"""获取所有结果"""
return self.results
def clear_results(self, older_than_days=7):
"""清理旧结果"""
cutoff_time = datetime.now().timestamp() - (older_than_days * 24 * 3600)
to_delete = []
for test_id, result in self.results.items():
end_time = result.get("end_time")
if end_time and end_time.timestamp() < cutoff_time:
to_delete.append(test_id)
for test_id in to_delete:
del self.results[test_id]
return len(to_delete)
def _emit_event(self, event_type, data):
"""触发事件（可扩展为事件总线）"""
# 这里可以实现事件监听器模式
# 例如：self.event_listeners[event_type].append(callback)
pass
def stop_all(self):
"""停止所有测试"""
self.executor.shutdown(wait=False)
self.running_tests.clear()# scenario_ui.py
from PyQt5.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QListWidget,
QListWidgetItem, QPushButton, QLabel, QGroupBox,
QTextEdit, QProgressBar, QTreeWidget, QTreeWidgetItem,
QSplitter, QTabWidget)
from PyQt5.QtCore import Qt, QTimer
class ScenarioUI(QWidget):
"""场景管理界面"""
def __init__(self, executor):
super().__init__()
self.executor = executor
self.current_scenario = None
self.init_ui()
self.load_scenarios()
self.setup_timer()
def init_ui(self):
"""初始化界面"""
layout = QHBoxLayout(self)
# 左侧：场景列表
left_panel = QWidget()
left_layout = QVBoxLayout(left_panel)
# 场景列表
self.scenario_list = QListWidget()
self.scenario_list.itemClicked.connect(self.on_scenario_selected)
left_layout.addWidget(QLabel("测试场景库"))
left_layout.addWidget(self.scenario_list)
# 场景描述
self.scenario_desc = QTextEdit()
self.scenario_desc.setReadOnly(True)
self.scenario_desc.setMaximumHeight(150)
left_layout.addWidget(QLabel("场景描述"))
left_layout.addWidget(self.scenario_desc)
# 执行按钮
self.execute_btn = QPushButton("执行选中场景")
self.execute_btn.clicked.connect(self.execute_selected_scenario)
self.execute_btn.setEnabled(False)
left_layout.addWidget(self.execute_btn)
# 右侧：执行控制面板
right_panel = QWidget()
right_layout = QVBoxLayout(right_panel)
# 当前执行状态
status_group = QGroupBox("执行状态")
status_layout = QVBoxLayout()
self.current_scenario_label = QLabel("无执行中的场景")
status_layout.addWidget(self.current_scenario_label)
self.progress_bar = QProgressBar()
status_layout.addWidget(self.progress_bar)
self.status_text = QTextEdit()
self.status_text.setReadOnly(True)
self.status_text.setMaximumHeight(100)
status_layout.addWidget(self.status_text)
self.stop_btn = QPushButton("停止当前测试")
self.stop_btn.clicked.connect(self.stop_current_test)
self.stop_btn.setEnabled(False)
status_layout.addWidget(self.stop_btn)
status_group.setLayout(status_layout)
right_layout.addWidget(status_group)
# 结果查看
results_group = QGroupBox("测试结果")
results_layout = QVBoxLayout()
self.results_tree = QTreeWidget()
self.results_tree.setHeaderLabels(["测试", "状态", "带宽", "延迟", "丢包"])
results_layout.addWidget(self.results_tree)
results_group.setLayout(results_layout)
right_layout.addWidget(results_group)
# 添加到主布局
splitter = QSplitter(Qt.Horizontal)
splitter.addWidget(left_panel)
splitter.addWidget(right_panel)
splitter.setSizes([300, 500])
layout.addWidget(splitter)
def load_scenarios(self):
"""加载场景列表"""
from test_scenarios import TestScenarioLibrary
scenarios = TestScenarioLibrary.list_scenarios()
for scenario in scenarios:
item = QListWidgetItem(scenario["name"])
item.setData(Qt.UserRole, scenario["id"])
self.scenario_list.addItem(item)
def on_scenario_selected(self, item):
"""场景被选中"""
scenario_id = item.data(Qt.UserRole)
from test_scenarios import TestScenarioLibrary
scenario = TestScenarioLibrary.get_scenario(scenario_id)
if scenario:
self.current_scenario = scenario_id
desc = f"名称: {scenario['name']}\n\n"
desc += f"描述: {scenario['description']}\n\n"
desc += f"包含测试: {len(scenario.get('tests', []))} 个\n\n"
desc += "测试项目:\n"
for test in scenario.get("tests", []):
desc += f" • {test.get('name', '未命名')}\n"
self.scenario_desc.setText(desc)
self.execute_btn.setEnabled(True)
def execute_selected_scenario(self):
"""执行选中的场景"""
if not self.current_scenario:
return
# 禁用按钮
self.execute_btn.setEnabled(False)
self.stop_btn.setEnabled(True)
# 更新状态
from test_scenarios import TestScenarioLibrary
scenario = TestScenarioLibrary.get_scenario(self.current_scenario)
self.current_scenario_label.setText(f"执行中: {scenario['name']}")
self.progress_bar.setRange(0, len(scenario.get("tests", [])))
self.progress_bar.setValue(0)
# 清空结果树
self.results_tree.clear()
# 异步执行
test_id = self.executor.execute_scenario_async(
self.current_scenario,
callback=self.on_scenario_completed
)
self.current_test_id = test_id
def on_scenario_completed(self, result):
"""场景完成回调"""
# 在主线程中更新UI
QTimer.singleShot(0, lambda: self.update_scenario_results(result))
def update_scenario_results(self, result):
"""更新场景结果"""
self.execute_btn.setEnabled(True)
self.stop_btn.setEnabled(False)
if result.get("all_passed"):
self.current_scenario_label.setText(
f"完成: {result['scenario_name']} (全部通过)"
)
else:
self.current_scenario_label.setText(
f"完成: {result['scenario_name']} (有失败)"
)
# 更新进度条
self.progress_bar.setValue(self.progress_bar.maximum())
# 更新结果树
for test_result in result.get("results", []):
item = QTreeWidgetItem(self.results_tree)
item.setText(0, test_result.get("name", "未知"))
status = test_result.get("status", "unknown")
item.setText(1, status)
if status == "passed":
item.setForeground(1, Qt.darkGreen)
elif status == "failed":
item.setForeground(1, Qt.darkRed)
elif status == "error":
item.setForeground(1, Qt.darkYellow)
bandwidth = test_result.get("bandwidth", 0)
item.setText(2, f"{bandwidth:.1f} Mbps" if bandwidth else "N/A")
latency = test_result.get("latency", 0)
item.setText(3, f"{latency:.1f} ms" if latency else "N/A")
loss = test_result.get("loss", 0)
item.setText(4, f"{loss:.2f}%" if loss is not None else "N/A")
# 显示摘要
summary = result.get("summary", {})
status_text = f"测试完成于: {result.get('end_time')}\n"
status_text += f"总测试数: {summary.get('total_tests', 0)}\n"
status_text += f"通过率: {summary.get('pass_rate', 0):.1f}%\n"
status_text += f"平均带宽: {summary.get('avg_bandwidth', 0):.1f} Mbps\n"
status_text += f"平均延迟: {summary.get('avg_latency', 0):.1f} ms\n"
self.status_text.setText(status_text)
def stop_current_test(self):
"""停止当前测试"""
# 这里需要实现停止逻辑
# 由于测试是在线程中运行的，需要优雅停止
self.stop_btn.setEnabled(False)
self.current_scenario_label.setText("正在停止...")
def setup_timer(self):
"""设置定时器更新状态"""
self.update_timer = QTimer()
self.update_timer.timeout.connect(self.update_status)
self.update_timer.start(1000) # 每秒更新
def update_status(self):
"""更新状态"""
if hasattr(self, 'current_test_id'):
status = self.executor.get_test_status(self.current_test_id)
if status["status"] == "running":
# 更新进度等
pass主要特性：
- 完整参数支持：支持所有 iperf3 命令行参数
- 图形化界面：直观的PyQt5界面
- 多标签页设计：基础配置、高级选项、服务器选项、测试结果、历史记录
- 实时结果显示：测试过程中实时显示结果
- 配置管理：保存和加载测试配置
- 历史记录：记录历次测试结果
- 结果导出：支持导出为文本和CSV格式
技术特点：
- 使用 QThread 实现异步测试，避免界面卡顿
- JSON配置文件管理
- 支持TCP和UDP协议
- 完整的错误处理和用户反馈
新增功能：
- 批量测试支持：支持定义和执行测试序列
- 插件系统：可扩展的插件架构
- 自动化脚本：Python API接口
- 网络质量评分：自动评估网络质量
- 详细报告生成：HTML/PDF格式报告
优化改进：
- 性能优化：减少内存使用，提高界面响应
- 稳定性增强：改进错误处理和恢复机制
- 用户体验：更直观的界面布局和操作流程
专业功能：
- 场景测试库：预定义各种测试场景
- API服务器：RESTful API接口
- WebSocket支持：实时数据传输
- 监控集成：Prometheus和Grafana集成
- 企业级报告：定制化报告模板
高级特性：
- 分布式测试支持
- 网络拓扑发现
- 性能基准对比
- SLA合规检查
云环境优化：
- 云服务集成：AWS、Azure、GCP等云平台集成
- 容器化支持：Docker镜像和Kubernetes部署
- 微服务架构：模块化设计，易于扩展
- 多云测试：跨云平台性能对比
现代化特性：
- 响应式Web界面
- 移动端适配
- 实时协作功能
- 人工智能分析
所有版本的配置文件都保持向后兼容：
- v1.0.0 配置文件可以在所有后续版本中使用
- 新版本增加的配置项会有默认值
- 旧版本无法识别的配置项会被忽略
REST API保持版本兼容：
/api/v1/路径下的API保持稳定- 新增功能使用新版本路径
/api/v2/ - 废弃的API会有足够长的弃用期
历史记录和数据格式：
- 所有版本都可以读取旧版本的历史数据
- 数据迁移工具自动处理格式变化
- 导出格式保持稳定
步骤：
-
备份现有数据：
cp iperf3_gui_config.json iperf3_gui_config.json.backup cp history_data.csv history_data.csv.backup
-
安装新版本：
pip install --upgrade iperf3-gui # 或直接替换文件 -
运行迁移脚本（如果需要）：
python migrate_v1_to_v2.py
-
验证升级：
- 检查配置是否正确加载
- 测试基本功能
- 验证历史数据
推荐步骤：
-
环境准备：
# Python环境 python -m venv venv source venv/bin/activate # Linux/macOS # 或 venv\Scripts\activate # Windows # 安装依赖 pip install -r requirements.txt # 安装iperf3 # Windows: 下载并添加到PATH # Linux: sudo apt-get install iperf3 # macOS: brew install iperf3
-
首次运行：
python iperf_gui.py
-
基本配置：
- 设置常用服务器地址
- 配置默认测试参数
- 设置结果保存路径
常见升级问题：
-
配置文件不兼容：
错误：JSON解析错误 解决：删除配置文件，程序会创建新的默认配置 -
依赖包冲突：
错误：ImportError 或版本冲突 解决：创建新的虚拟环境，重新安装 -
历史数据丢失：
现象：历史记录为空 解决：检查备份文件，手动导入 -
性能下降：
现象：新版本运行变慢 解决：检查系统资源，调整配置参数
获取帮助：
- 查看本文档的故障排除章节
- 访问GitHub Issues页面
- 查阅在线文档
| 图形界面选项 | iperf3 参数 | 说明 |
|---|---|---|
| 测试模式 | -c / -s |
客户端/服务器模式 |
| 服务器地址 | -c <host> |
客户端模式必须 |
| 端口 | -p <port> |
默认5201 |
| 协议 | -u |
UDP协议 |
| 测试时间 | -t <seconds> |
测试持续时间 |
| 并行流 | -P <num> |
并行连接数 |
| 带宽限制 | -b <rate> |
带宽限制 |
| 窗口大小 | -w <size> |
TCP窗口大小 |
| MSS大小 | -M <size> |
TCP最大段大小 |
| 缓冲区长度 | -l <length> |
读写缓冲区长度 |
| 传输数据量 | -n <bytes> |
传输指定数据量 |
| 块数量 | -k <count> |
块数量 |
| 报告间隔 | -i <seconds> |
报告间隔 |
| 格式 | --format <unit> |
输出单位格式 |
| 反向测试 | -R |
反向传输 |
| 双向测试 | --bidir |
双向同时传输 |
| 连接超时 | --connect-timeout <ms> |
连接超时 |
| 绑定设备 | --bind-dev <dev> |
绑定网络设备 |
| 绑定主机 | -B <host> |
绑定源地址 |
| 客户端端口 | --cport <port> |
客户端端口 |
| 省略时间 | -O <seconds> |
跳过开始时间 |
| 标题 | -T <title> |
输出标题 |
| 额外数据 | --extra-data <str> |
JSON额外数据 |
| 获取服务器输出 | --get-server-output |
获取服务器输出 |
| UDP 64位计数器 | --udp-counters-64bit |
UDP使用64位计数器 |
| 重复负载 | --repeating-payload |
UDP重复负载 |
| 不分片 | --dont-fragment |
设置DF标志 |
| IPv4 only | -4 |
仅使用IPv4 |
| IPv6 only | -6 |
仅使用IPv6 |
| TOS | -S <tos> |
IP服务类型 |
| DSCP | --dscp <dscp> |
差分服务代码点 |
| 禁用Nagle | -N |
禁用Nagle算法 |
| 零拷贝 | -Z |
使用零拷贝 |
| 跳过接收拷贝 | --skip-rx-copy |
接收端零拷贝 |
| 接收超时 | --rcv-timeout <ms> |
接收超时 |
| 用户名 | --username <name> |
认证用户名 |
| RSA公钥路径 | --rsa-public-key-path <path> |
RSA公钥文件 |
| 文件传输 | -F <filename> |
发送文件内容 |
| 单次连接 | -1 |
服务器单次连接 |
| 守护进程 | -D |
守护进程模式 |
| 服务器带宽限制 | --server-bitrate-limit <rate> |
服务器带宽限制 |
| 空闲超时 | --idle-timeout <seconds> |
空闲超时 |
| 服务器最大持续时间 | --server-max-duration <seconds> |
服务器运行时间限制 |
| RSA私钥路径 | --rsa-private-key-path <path> |
RSA私钥文件 |
| 授权用户路径 | --authorized-users-path <path> |
用户配置文件 |
| 时间偏差阈值 | --time-skew-threshold <seconds> |
时间偏差阈值 |
| 使用PKCS1填充 | --use-pkcs1-padding |
RSA使用PKCS1填充 |
| PID文件 | -I <pidfile> |
写入PID文件 |
| 日志文件 | --logfile <filename> |
日志文件 |
| JSON输出 | -J |
JSON格式输出 |
| JSON流输出 | --json-stream |
JSON流输出 |
| 完整JSON流输出 | --json-stream-full-output |
完整JSON流 |
| 详细输出 | -V |
详细输出 |
| 调试模式 | -d 或 --debug=<level> |
调试模式 |
| 显示版本 | -v |
显示版本 |
| 强制刷新 | --forceflush |
强制刷新输出 |
| 时间戳 | --timestamps[=format] |
时间戳 |
| CPU亲和性 | -A <cpuid> |
CPU亲和性 |
- 10M以太网：理论 10 Mbps，实际 9-9.5 Mbps
- 100M以太网：理论 100 Mbps，实际 94-98 Mbps
- 千兆以太网：理论 1000 Mbps，实际 940-980 Mbps
- 万兆以太网：理论 10000 Mbps，实际 9400-9800 Mbps
-
Wi-Fi 4 (802.11n)：
- 2.4GHz：理论 150-300 Mbps，实际 50-150 Mbps
- 5GHz：理论 300-450 Mbps，实际 100-250 Mbps
-
Wi-Fi 5 (802.11ac)：
- 理论 433-1733 Mbps，实际 200-800 Mbps
- 受距离和障碍物影响大
-
Wi-Fi 6 (802.11ax)：
- 理论 600-9608 Mbps，实际 400-1200 Mbps
- 多设备性能更好
- ADSL：下行 8-24 Mbps，上行 1-3 Mbps
- VDSL：下行 50-100 Mbps，上行 10-20 Mbps
- 光纤到户：
- 家庭：100-1000 Mbps，上下行对称或不对称
- 企业：100-10000 Mbps，通常对称
- 4G LTE：
- 理论 100-300 Mbps，实际 20-80 Mbps
- 受信号强度和用户数影响
- 5G：
- 理论 1-10 Gbps，实际 100-1000 Mbps
- 毫米波可达更高速度
- 带宽 (Bandwidth)：网络传输数据的能力，单位 bps
- 吞吐量 (Throughput)：实际达到的数据传输速率
- 线速 (Line Rate)：理论最大传输速率
- 有效带宽 (Effective Bandwidth)：考虑协议开销后的实际带宽
- 延迟 (Latency)：数据从发送到接收的时间
- 往返时间 (RTT)：数据往返所需时间
- 抖动 (Jitter)：延迟的变化程度
- 单向延迟 (One-way Delay)：单向传输时间
- 丢包率 (Packet Loss Rate)：丢失的数据包比例
- 乱序 (Out-of-Order)：数据包到达顺序错误
- 重复包 (Duplicate Packet)：重复接收的数据包
- 错误率 (Error Rate)：包含错误的数据包比例
- 重传 (Retransmission)：TCP重新发送丢失的数据
- 拥塞窗口 (Congestion Window)：TCP根据网络状况调整的窗口
- 慢启动 (Slow Start)：TCP初始阶段指数增长窗口
- 快速重传 (Fast Retransmit)：收到3个重复ACK立即重传
{
"name": "快速网络检查",
"server": "目标IP",
"port": 5201,
"duration": 10,
"parallel": 4,
"protocol": "tcp",
"interval": 1,
"expected_min_bandwidth": 500,
"expected_max_latency": 50,
"expected_max_loss": 0.1
}{
"name": "全面网络评估",
"tests": [
{
"name": "TCP单流基准",
"protocol": "tcp",
"parallel": 1,
"duration": 30
},
{
"name": "TCP多流性能",
"protocol": "tcp",
"parallel": 8,
"duration": 60
},
{
"name": "UDP质量测试",
"protocol": "udp",
"duration": 30,
"bandwidth": "100M",
"check": ["loss", "jitter"]
},
{
"name": "双向压力测试",
"protocol": "tcp",
"parallel": 16,
"duration": 300,
"bidirectional": true
}
],
"schedule": "每月执行",
"report": "详细报告"
}{
"name": "SLA符合性验证",
"sla_requirements": {
"bandwidth": "≥ 95% of 1000 Mbps",
"latency": "≤ 50 ms",
"loss": "≤ 0.1%",
"availability": "≥ 99.9%"
},
"test_frequency": "每小时",
"test_duration": 300,
"alert_thresholds": {
"warning": "80% of SLA",
"critical": "60% of SLA"
},
"reporting": "实时仪表板 + 月度报告"
}- ECONNREFUSED (111)：连接被拒绝，检查服务器和防火墙
- ETIMEDOUT (110)：连接超时，检查网络连通性
- EHOSTUNREACH (113)：主机不可达，检查路由
- ENETUNREACH (101)：网络不可达，检查网络配置
- 低带宽：
- 检查网卡协商状态
- 检查中间设备限制
- 检查系统资源使用
- 高延迟：
- 使用 traceroute 分析路径
- 检查 QoS 配置
- 检查网络拥塞
- 高丢包：
- 检查线缆质量
- 检查设备错误计数
- 检查 MTU 设置
- iperf3 未找到：安装 iperf3 或检查 PATH
- 权限不足：使用 sudo 或管理员权限
- 端口被占用：更改端口或停止占用程序
- 内存不足：减少并行流或测试时间
- iperf3 官网：https://software.es.net/iperf/
- 源代码：https://github.com/esnet/iperf
- 文档：https://iperf.fr/iperf-doc.php
- 网络性能测试指南：
- https://www.ietf.org/rfc/rfc2544.txt (基准测试方法学)
- https://www.ietf.org/rfc/rfc6349.txt (TCP测试考虑)
- TCP/IP 详解：经典网络技术书籍
- 网络诊断工具大全：学习各种网络工具的使用
- GitHub Issues：报告问题和功能请求
- Stack Overflow：技术问题讨论
- 网络技术论坛：专业交流和学习
- ping：基本连通性测试
- traceroute：路径追踪
- mtr：结合 ping 和 traceroute
- tcpdump/Wireshark：抓包分析
- netperf：另一个网络性能测试工具
- iPerf2：旧版本，有些场景仍在使用
iperf3 图形化工具不仅仅是一个简单的界面包装，它是：
- 学习工具：帮助理解网络性能测试原理
- 诊断工具：快速定位网络问题
- 验证工具：确保网络服务符合预期
- 规划工具：为网络扩容提供数据支持
- 监控工具：持续跟踪网络质量变化
- 从简单开始：先使用默认配置，了解基本功能
- 逐步深入：根据需要探索高级选项
- 记录实践：记录每次测试的环境和结果
- 持续学习：网络技术不断发展，保持学习
- 分享经验：在社区分享使用经验和技巧
工具会持续更新和改进，计划中的功能包括：
- AI分析：智能分析测试结果，自动给出建议
- 云集成：深度集成云服务商的监控和测试功能
- 移动端：开发手机App，随时随地进行测试
- 协作功能：多人协作测试和分析
- 更多协议：支持HTTP/3、QUIC等新协议
感谢所有使用、测试和贡献的用户，特别是：
- 早期测试用户提供的宝贵反馈
- GitHub社区的issue报告和讨论
- 开源项目的贡献者们
- 网络技术社区的分享和交流
虽然我们不提供私人邮箱支持，但您可以通过以下方式参与：
- GitHub Issues：报告bug和请求功能
- Pull Requests：贡献代码和改进
- 文档贡献：帮助完善文档
- 社区讨论：分享使用经验
记住：最好的支持是公开的支持，您的问题和解决方案可以帮助更多人。
版权声明 © 永久 杜玛 保留所有权利
最后更新：2024年1月 文档版本：v1.0 工具版本：iperf3 GUI v1.0.0
注意：本文档内容会随着工具更新而更新，请定期查看最新版本。
免责声明：本文档提供的信息仅供参考，作者不对因使用本文档或相关工具造成的任何损失负责。网络测试可能影响网络性能，请在适当的环境下进行测试。