Python Modbus高频数据记录脚本异常：长时间运行后停止采集

针对你遇到的高频采集（6-9ms/次）脚本长时间运行后静默停止、CSV数据不足甚至为空的问题，结合你的环境（Windows10 + Python3.9 + pymodbus/pandas/matplotlib），可以从以下几个方向排查解决：

一、内存溢出优化 #

持续向单个pandas DataFrame追加数据会导致内存占用线性增长，长时间运行后可能触发系统隐性内存回收或进程终止，且无报错提示。

• 改用分块写入CSV：每采集固定条数（比如1000条）就将临时数据写入CSV，随后清空临时DataFrame释放内存。第一次写入时带表头，后续追加跳过表头：

  import pandas as pd
  import os
  
  csv_path = "data.csv"
  temp_data = []
  batch_size = 1000
  
  # 采集到单条数据后加入临时列表
  temp_data.append({"RPM": rpm_val, "扭矩": torque_val, ...})
  
  # 达到批次大小则写入
  if len(temp_data) >= batch_size:
      df = pd.DataFrame(temp_data)
      df.to_csv(csv_path, mode='a', header=not os.path.exists(csv_path), index=False)
      temp_data.clear()  # 清空临时数据
  

• 每次清空临时数据后，手动触发垃圾回收：

  import gc
  gc.collect()
  

二、Modbus连接稳定性修复 #

高频通信下，连接可能隐性断开或请求超时，pymodbus默认逻辑不会主动报错，导致采集静默停止。

• 每次采集前检查连接状态，断开则自动重连：

  from pymodbus.client import ModbusTcpClient
  
  client = ModbusTcpClient("plc_ip", port=502)
  
  def ensure_connection():
      if not client.is_socket_open():
          try:
              client.connect()
              print("重新建立Modbus连接")
          except Exception as e:
              print(f"重连失败: {str(e)}")
  

• 给Modbus请求设置匹配采集频率的超时时间，避免请求挂起阻塞主线程：

  # 超时设为10ms，略高于单次采集耗时
  response = client.read_holding_registers(start_addr, register_count, timeout=0.01)
  

• 捕获Modbus通信异常，避免静默失败：

  from pymodbus.exceptions import ModbusIOException
  
  try:
      response = client.read_holding_registers(...)
      if response.isError():
          print(f"Modbus响应错误: {response}")
          ensure_connection()
  except ModbusIOException as e:
      print(f"通信异常: {str(e)}")
      ensure_connection()
  

三、Windows系统进程限制排查 #

Windows对后台进程的资源调度可能存在隐性限制，导致高频任务被系统终止但无控制台报错。

• 将脚本进程优先级设为高，避免被系统调度压制：

  import psutil
  
  proc = psutil.Process()
  proc.nice(psutil.HIGH_PRIORITY_CLASS)  # Windows环境下生效
  

• 打开Windows事件查看器（eventvwr.msc），在「Windows日志-应用程序」中查找脚本进程的终止日志，确认是否被系统强制结束。

四、CSV写入可靠性提升 #

写入CSV时的缓冲区未刷新、磁盘异常或文件被占用，可能导致数据丢失或仅存表头。

• 写入时强制刷新磁盘缓冲区，避免数据停留在内存中：

  import os
  
  with open(csv_path, 'a', newline='') as f:
      df.to_csv(f, header=not os.path.exists(csv_path), index=False)
      f.flush()
      os.fsync(f.fileno())  # 强制将缓冲区数据写入磁盘
  

• 使用临时文件写入，完成后替换目标文件，避免中途崩溃导致文件损坏：

  import shutil
  
  temp_csv = "temp_data.csv"
  # 先写入临时文件
  df.to_csv(temp_csv, mode='a', header=not os.path.exists(temp_csv), index=False)
  # 替换目标文件
  shutil.move(temp_csv, csv_path)
  

五、主线程阻塞排查 #

若脚本包含matplotlib实时绘图逻辑，高频采集时绘图操作可能阻塞主线程，导致采集停止。

• 将绘图逻辑放到独立线程中，通过队列传递数据，不影响采集主线程：

  import threading
  import queue
  
  data_queue = queue.Queue()
  
  def plot_task():
      while True:
          if not data_queue.empty():
              data = data_queue.get()
              # 执行绘图逻辑，比如更新图表
              # ...
  
  # 启动绘图线程（设为守护线程，随主线程退出）
  plot_thread = threading.Thread(target=plot_task, daemon=True)
  plot_thread.start()
  
  # 采集到数据后放入队列
  data_queue.put(new_data_dict)
  

内容的提问来源于stack exchange，提问作者Arie3s