如何获取进程的每秒读写字节数

嘿，这个需求我之前做性能监控的时候刚好折腾过！其实原理和pidstat -dl完全一致——通过两次采样进程的IO计数器，计算时间间隔内的增量，再除以时间差，就能得到每秒的读写字节数，毕竟psutil.io_counters()返回的是进程从启动到当前的累计读写量嘛。

我给你整理了一个实用的代码示例，直接就能用：

import psutil
import time
from psutil import NoSuchProcess

def monitor_process_io_per_sec(pid, interval=1, max_duration=None):
    # 先校验目标进程是否存在
    try:
        target_process = psutil.Process(pid)
    except NoSuchProcess:
        print(f"错误：PID {pid} 对应的进程不存在！")
        return

    start_monitor_time = time.time()
    # 记录第一次的IO数据和时间戳
    prev_io_data = target_process.io_counters()
    prev_timestamp = start_monitor_time

    print(f"开始监控 PID {pid} 的每秒IO情况（按Ctrl+C停止）...")
    try:
        while True:
            # 检查是否达到预设的最大监控时长
            if max_duration and (time.time() - start_monitor_time) >= max_duration:
                print(f"\n已达到设置的最大监控时长 {max_duration} 秒，监控结束")
                break

            # 等待指定的采样间隔
            time.sleep(interval)
            curr_timestamp = time.time()
            try:
                # 获取当前时刻的IO计数器数据
                curr_io_data = target_process.io_counters()
            except NoSuchProcess:
                print(f"\nPID {pid} 进程已终止，监控结束")
                break

            # 计算实际流逝的时间（避免sleep的微小精度误差）
            time_elapsed = curr_timestamp - prev_timestamp
            # 换算成每秒读写字节数
            read_per_sec = (curr_io_data.read_bytes - prev_io_data.read_bytes) / time_elapsed
            write_per_sec = (curr_io_data.write_bytes - prev_io_data.write_bytes) / time_elapsed

            # 格式化输出，带时间戳更直观
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 每秒读: {read_per_sec:.2f} bytes/s | 每秒写: {write_per_sec:.2f} bytes/s")

            # 更新上一次的采样基准
            prev_io_data = curr_io_data
            prev_timestamp = curr_timestamp

    except KeyboardInterrupt:
        print("\n用户手动停止监控")

# 调用示例：替换成你要监控的PID，比如监控PID 1234，每秒采样一次，持续20秒
# monitor_process_io_per_sec(pid=1234, interval=1, max_duration=20)

小细节补充 #

• 异常处理：考虑了进程不存在、进程中途退出的情况，也支持手动按Ctrl+C停止监控，或者设置最大时长自动终止。
• 精度优化：没有硬用设置的interval作为时间差，而是用实际的curr_timestamp - prev_timestamp，避免time.sleep()的微小延迟影响结果准确性。
• 如果只需要单次计算而非持续监控，也可以简化成这样：

  import psutil
  import time
  
  def calculate_io_per_sec(pid, interval=1):
      process = psutil.Process(pid)
      prev_io = process.io_counters()
      time.sleep(interval)
      curr_io = process.io_counters()
      return {
          "read_per_sec": (curr_io.read_bytes - prev_io.read_bytes) / interval,
          "write_per_sec": (curr_io.write_bytes - prev_io.write_bytes) / interval
      }
  
  # 调用示例
  io_stats = calculate_io_per_sec(1234)
  print(f"每秒读: {io_stats['read_per_sec']:.2f} bytes/s, 每秒写: {io_stats['write_per_sec']:.2f} bytes/s")
  

这样就能灵活满足不同的监控需求啦！

备注：内容来源于stack exchange，提问作者M1hv771