Python代码为何为同一任务创建两个Worker线程？

核心问题原因 #

• Flask开发模式的多进程实例隔离
  Flask默认debug模式会启动两个进程（主进程+重载器），全局变量dev = devices()会在每个进程中独立初始化。当请求触发时，两个进程都会执行add_device逻辑，各自创建Worker线程并加入自身的workers字典，最终导致同一IP对应两个独立运行的线程。停止线程时，仅能修改其中一个进程内的Worker实例状态，另一个进程的线程不受影响。

• 接口中不必要的join()调用
  add_device里的join()会阻塞Flask的请求处理线程，直到Worker线程终止，但Worker是无限循环逻辑，这会导致请求长期挂起。同时在多进程场景下，两个进程都会执行这段代码，进一步加剧重复创建线程的问题。

• 线程创建前缺少重复检查
  未判断当前IP是否已存在对应的Worker线程，即使同一IP被多次请求，也会重复创建新线程，叠加多进程问题后，线程数量翻倍。

排查与修复建议 #

• 关闭Flask开发模式的重载功能
  运行Flask时禁用重载器，确保仅启动一个进程：

  server.run(debug=True, use_reloader=False)
  

  或通过环境变量设置：FLASK_DEBUG=1 FLASK_RUN_USE_RELOADER=false flask run

• 移除join()并添加重复线程检查
  修改add_device逻辑，让Worker后台运行，同时避免重复创建：

  @server.route('/add_device/',methods=['POST'])
  def add_device():
      data = request.get_json()
      ip_address = data["IP"]
      interface = data["Interface"]
      
      # 检查是否已有该IP的线程，避免重复创建
      if ip_address not in dev.get_workers():
          thread = Worker(ip_address, interface)
          dev.update_workers(thread, ip_address)
          thread.start()
      return 200
  

• 修复devices类的参数拼写错误
  将update_workers方法的参数threa改为thread，避免后续维护混淆：

  def update_workers(self, thread, ip_address):
      self.workers[ip_address] = thread
  

• 优化线程终止的可靠性
  改用threading.Event替代布尔变量，解决线程阻塞在requests或sleep时无法及时退出的问题：

  import threading
  import time
  import requests
  
  class Worker(threading.Thread):
      def __init__(self, ip_address, interface):
          super().__init__()
          self.ip_address = ip_address
          self.interface = interface
          self.stop_event = threading.Event()
  
      def setExit(self):
          self.stop_event.set()
          print("Stop signal received")
  
      def run(self):
          while not self.stop_event.is_set():
              try:
                  print(f"Thread ID: {threading.get_ident()}")
                  # 给请求添加超时，避免长时间阻塞
                  response = requests.get(
                      f'https://{self.ip_address}:5001/node_util/all/{self.interface}',
                      verify=False,
                      timeout=5
                  )
                  response.json()
                  # 使用Event.wait替代sleep，响应停止信号
                  self.stop_event.wait(10)
              except Exception as e:
                  print(f"Error occurred: {e}")
                  # 异常后短暂等待，同时检查停止信号
                  if self.stop_event.wait(1):
                      break
  

• 多进程部署下的状态共享方案（可选）
  如果需要用多进程模式运行Flask，需使用进程间共享存储（如multiprocessing.Manager）或外部缓存（如Redis）管理线程状态，确保所有进程操作统一的设备线程字典。

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