如何在Linux上运行最优数量的单个Python脚本实例

先明确下你的核心场景：

我有一个脚本，可对约1200个不同文件执行独立任务。它会遍历每个文件，检查该文件是否已完成处理或正在处理中；若未完成且未被处理（非并行运行时不会出现此状态），则对该文件执行任务。大致流程如下：myScript.py：doTask()在我的机器上耗时20-40分钟，内存需求从初始的极小值增长至约8GB…

结合这个场景，我从内存控制、效率提升、稳定性保障三个核心方向给你具体的落地建议：

一、先搞定内存暴涨的问题 #

从初始极小值涨到8GB，大概率是doTask()存在内存泄漏，或者没有及时释放大对象。给你两个实用方案：

• 手动清理+强制垃圾回收：每次处理完一个文件后，显式删除任务中用到的大对象（比如加载的文件内容、大型数据集），再触发垃圾回收，避免内存累积。示例：

  def doTask(file_path):
      # 你的任务逻辑
      large_data = load_large_file(file_path)
      process(large_data)
      # 处理完立即清理资源
      del large_data
      import gc
      gc.collect()
  

• 用工具定位泄漏点：如果不知道哪里出了问题，用tracemalloc做内存快照对比，精准找出内存占用最高的代码行。示例代码：

  import tracemalloc
  
  tracemalloc.start()
  # 先拿一个测试文件跑一次任务
  doTask(test_file_path)
  snapshot = tracemalloc.take_snapshot()
  top_stats = snapshot.statistics('lineno')
  
  print("内存占用Top 10的代码行：")
  for stat in top_stats[:10]:
      print(stat)
  

  根据输出就能针对性优化未释放的对象。

二、并行化把总耗时砍下来 #

单线程处理1200个文件，总耗时最少400小时（1200*20分钟），效率太低。因为每个文件任务完全独立，非常适合用多进程并行处理（多线程受GIL限制，内存占用高的任务用进程更合适）：

• 进程池实现并行+状态防冲突：用multiprocessing.Pool，根据你的机器内存设置进程数（比如你有64GB内存，每个任务占8GB，就设8个进程）。注意多进程下要做好状态的原子性检查，避免多个进程抢同一个文件：

  from multiprocessing import Pool
  import portalocker  # 用文件锁确保状态操作的原子性
  import os
  
  def process_single_file(file_path):
      # 用文件锁避免多个进程同时修改状态
      lock_path = f"{file_path}.lock"
      with open(lock_path, "w") as lock_file:
          try:
              portalocker.lock(lock_file, portalocker.LOCK_EX | portalocker.LOCK_NB)
              # 检查文件状态
              if os.path.exists(f"{file_path}.done"):
                  return
              # 标记正在处理
              open(f"{file_path}.processing", "w").close()
          except portalocker.LockException:
              # 其他进程在处理，直接跳过
              return
  
      try:
          doTask(file_path)
          # 标记处理完成
          open(f"{file_path}.done", "w").close()
      finally:
          # 清理正在处理的标记
          if os.path.exists(f"{file_path}.processing"):
              os.remove(f"{file_path}.processing")
          # 释放锁
          os.remove(lock_path)
  
  if __name__ == "__main__":
      file_list = get_your_file_list()  # 替换成你的1200个文件路径列表
      with Pool(processes=8) as pool:
          pool.map(process_single_file, file_list)
  

  这里用.processing、.done状态文件+文件锁，确保多进程下不会重复处理或冲突。

三、断点续跑+监控，避免白忙活 #

1200个任务跑几天很容易中途出问题，加断点续跑和监控能大幅提升稳定性：

• 持久化状态：把每个文件的处理状态存在本地文件或SQLite数据库里，脚本重启后直接读取状态，跳过已完成的任务，不用从头再来
• 加日志记录：每个任务的开始、结束、结果都记到日志里，方便排查问题。示例：

  import logging
  logging.basicConfig(filename='task_log.log', level=logging.INFO, format='%(asctime)s - %(message)s')
  
  # 在process_single_file中添加
  logging.info(f"开始处理文件: {file_path}")
  # 处理完成后
  logging.info(f"文件处理完成: {file_path}")
  # 异常时
  logging.error(f"文件处理失败 {file_path}: {str(e)}")
  

• 资源监控：用psutil模块监控进程内存，防止内存溢出：

  import psutil
  import os
  
  def doTask(file_path):
      process = psutil.Process(os.getpid())
      # 你的任务逻辑
      # 中途检查内存占用
      mem_usage = process.memory_info().rss / 1024 / 1024 / 1024  # 转成GB
      if mem_usage > 7.5:  # 设个安全阈值，比如7.5GB
          raise MemoryError("内存占用超过阈值，终止任务")
  

额外小技巧：单任务内存优化 #

如果doTask()处理的是大文件，尽量分块读取，不要一次性加载整个文件到内存。比如处理CSV用pandas的chunksize参数，处理文本文件用迭代器逐行读取，从源头降低内存占用。

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