使用Python多线程时如何防止输出内容重叠(overprinting)问题

根本原因 #

Python标准库的print()方法没有内置线程安全保障，多个线程同时调用print()时，不同线程写入标准输出的字符流会出现穿插、拼接，最终导致视觉上的输出重叠。前15行输出正常是因为程序启动初期只有少量线程在运行，不存在并发打印的场景，当所有线程都启动进入运行态后，并发打印就会触发冲突。

可直接落地的修复方法 #

• 方案1：全局打印锁（改动最小，适合中小规模项目）
  新增线程锁包裹所有打印操作，保证同一时间只有一个线程能向标准输出写内容，完全避免并发冲突。
  参考实现：

  import threading
  
  # 全局唯一打印锁
  print_lock = threading.Lock()
  
  # 替换所有原生print调用
  def safe_print(*args, **kwargs):
      with print_lock:
          print(*args, **kwargs)
  
  # 线程函数内使用示例
  def worker_thread(task_id):
      # 原print(f"处理任务{task_id}完成") 修改为：
      safe_print(f"处理任务{task_id}完成")
  

• 方案2：队列中转打印（性能更优，适合高并发多线程场景）
  所有线程将待打印内容提交到队列，单独启动一个专用线程负责从队列拉取内容打印，从根源上规避多线程写标准输出的冲突问题，高并发下性能优于锁方案。
  参考实现：

  import threading
  from queue import Queue
  
  print_queue = Queue()
  
  # 专用打印线程
  def print_daemon():
      while True:
          msg = print_queue.get()
          if msg == "__EXIT__":
              print_queue.task_done()
              break
          print(msg)
          print_queue.task_done()
  
  # 程序启动时先启动打印线程
  threading.Thread(target=print_daemon, daemon=True).start()
  
  # 线程函数内使用示例
  def worker_thread(task_id):
      print_queue.put(f"处理任务{task_id}完成")
  
  # 所有任务执行完成后，通知打印线程退出并等待剩余内容打印完成
  # print_queue.put("__EXIT__")
  # print_queue.join()
  

两种方案都可以完全解决输出重叠的问题，可根据项目规模和并发量选择。

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