多进程目录扫描脚本异常：部分远程/本地目录停止扫描求助

看起来你遇到的是多进程环境下远程连接静默失效导致的扫描停滞问题——进程没终止，但就是不再列出目录里的新文件，重启服务又能恢复。结合你的代码和场景，我整理了可能的原因和对应的解决思路：

可能的故障原因 #

• 远程连接静默断开，未触发异常：大部分FTP/SFTP/SMB服务器会对长时间无活动的连接自动断开，但有些客户端库不会主动检测这种断开状态。比如你的代码里用ftp.dir()获取目录列表，如果连接已经被服务器静默断开，这个调用可能会卡住（底层socket处于半开状态，没有报错信号），或者返回空列表，导致程序误以为目录里没有文件。而单进程的脚本因为访问频率稳定，连接不会被服务器判定为闲置，所以没出现问题。
• 远程操作无超时限制：你的代码里没有给FTP/SFTP/SMB的操作设置超时时间，一旦服务器响应缓慢或者连接出现网络波动，listdir里的远程调用可能会无限期阻塞，进程就停在这一步不再继续执行。
• 多进程环境下的连接资源管理问题：虽然每个进程都有独立的连接实例，但某些底层库在多进程环境下可能存在全局资源泄漏（比如未正确释放的socket句柄），或者服务器端对同一客户端IP的并发连接数有限制，导致部分连接被静默拒绝。

解决思路 #

1. 添加连接健康检查与自动重连机制 #

在每次执行目录扫描前，先检查连接是否有效，失效则自动重建连接：

• 给每个连接类添加健康检查方法，比如FTP的：

  def check_connection(self):
      try:
          # 发送NOOP命令，FTP服务器会响应这个空操作，用于验证连接
          self.ftp.voidcmd('NOOP')
          return True
      except (ftplib.error_temp, ftplib.error_perm, socket.error, ConnectionResetError):
          # 捕获连接相关异常，判定连接失效
          return False
  

• 主循环中加入检查逻辑：

  import time
  while True:
      # 先检查连接是否有效
      if not connect.check_connection():
          logging.warning(f"Connection to {self.source['host']} lost, reconnecting...")
          # 重建连接
          connect = Libs.connection.Connection(self.config, server)
      try:
          files = connect.listdir()
          # 处理文件逻辑...
      except Exception as e:
          logging.error(f"Scan failed: {str(e)}, will reconnect next cycle")
      # 添加短睡眠，避免频繁访问服务器
      time.sleep(5)
  

2. 给远程操作设置超时时间 #

避免远程调用无限期阻塞：

• 在初始化FTP连接时添加超时参数：

  # 连接时设置30秒超时，可根据实际情况调整
  self.ftp.connect(source, self.source['port'], timeout=30)
  

• 对于不支持直接设置超时的操作（比如部分SFTP/SMB方法），可以用线程包装操作并实现超时控制：

  import threading
  def timed_listdir(self, dirftp=None, timeout=30):
      result = []
      exception = None
      def target():
          nonlocal result, exception
          try:
              result = self._original_listdir(dirftp)
          except Exception as e:
              exception = e
      thread = threading.Thread(target=target)
      thread.start()
      thread.join(timeout)
      if thread.is_alive():
          raise TimeoutError("Listdir operation timed out")
      if exception:
          raise exception
      return result
  

  然后把原来的listdir改名为_original_listdir，对外暴露timed_listdir方法。

3. 增强异常捕获与日志记录 #

在扫描逻辑中添加全面的异常捕获，方便定位问题：

• 修改listdir方法，捕获所有可能的远程操作异常：

  def listdir(self, dirftp = None):
      lines, filesarr, folders = [], [], []
      try:
          if dirftp is None:
              self.ftp.dir(lines.append)
          else:
              self.ftp.dir(dirftp, lines.append)
          # 原有解析逻辑...
          return filesarr
      except Exception as e:
          logging.error(f"Failed to list directory {dirftp}: {str(e)}")
          # 抛出异常，让上层触发重连
          raise
  

• 记录详细的日志，包括连接状态、扫描结果、异常信息，方便后续排查。

4. 检查并发连接限制 #

• 查看远程服务器的日志，确认是否存在并发连接数限制，导致部分连接被拒绝。如果有，可以调整进程数量，或者在连接失败时添加重试逻辑，避免一次性建立过多连接。
• 检查操作系统的socket资源限制（比如ulimit -n），确保有足够的资源支持11个进程的并发连接。

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