编写一个使用 Python 进行 FIO 测试的代码示例，可以通过调用 `fio` 工具的命令行接口来实现。Python 可以使用 `subprocess` 模块运行系统命令并捕获输出结果，从而实现对磁盘性能的自动化测试。 以下是一个简单的 Python 示例代码，用于运行 FIO 测试并解析其输出： ```python import subprocess import json def run_fio_test(): # 定义 FIO 测试参数 fio_command = [ 'fio', '--name=randread', # 测试名称 '--ioengine=libaio', # I/O 引擎 '--direct=1', # 使用直接 I/O，绕过缓存 '--rw=randread', # 随机读取测试 '--bs=4k', # 块大小为 4KB '--size=512M', # 测试文件大小为 512MB '--numjobs=4', # 并发任务数为 4 '--runtime=60', # 运行时间为 60 秒 '--time_based', # 基于时间的测试 '--group_reporting', # 分组报告 '--output-format=json' # 输出格式为 JSON ] # 执行 FIO 命令并捕获输出 result = subprocess.run(fio_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode == 0: output = result.stdout.decode('utf-8') try: # 将输出解析为 JSON 格式 fio_output = json.loads(output) # 提取随机读取测试的结果 read_results = fio_output['jobs'][0]['read'] print("FIO Test Results:") print(f"IOPS: {read_results['iops']}") print(f"Bandwidth (KB/s): {read_results['bw']}") print(f"Latency (ms): {read_results['lat_ns']['mean'] / 1000000}") except Exception as e: print("Failed to parse FIO output:", e) else: print("FIO command failed with error:", result.stderr.decode('utf-8')) if __name__ == "__main__": run_fio_test() ``` ### 解释： 1. **FIO 测试参数**：该脚本定义了一组常见的 FIO 参数，进行随机读取测试（`--rw=randread`），使用 `libaio` 作为 I/O 引擎（`--ioengine=libaio`），并且启用了直接 I/O（`--direct=1`）。 2. **子进程执行**：通过 `subprocess.run()` 方法执行 FIO 命令，并捕获标准输出和错误信息。 3. **JSON 解析**：FIO 支持将输出格式化为 JSON（`--output-format=json`），这样可以方便地解析结果。 4. **结果提取**：从 JSON 输出中提取关键指标，如 IOPS、带宽（BW）和延迟（Latency），并打印到控制台。 ### 注意事项： - 确保在运行此脚本之前已经安装了 `fio` 工具，并且可以在命令行中直接调用。 - 如果需要更复杂的测试场景，可以根据需求调整 FIO 参数。 - 此示例仅展示了随机读取测试，实际应用中可以根据需要修改为顺序读写或其他类型的测试 [^3]。