注释print语句后Multiprocessing代码运行结果异常的原因

我编写了一段使用Python multiprocessing模块的Process和Queue的代码：

from random import random
from multiprocessing import Process
from multiprocessing import Queue
import time


def new(shared_queue):
    print('Consumer: Running', flush=True)
    while shared_queue.qsize() > 0:
        shared_queue.get()
        print(shared_queue.qsize())
    print('Consumer: Done', flush=True)


if __name__ == '__main__':
    start_time = time.time()

    # 初始化队列
    queue = Queue()
    for _ in range(50000):
        value = random()
        queue.put(value)

    print('Init Queue size', queue.qsize())

    p0 = Process(target=new, args=(queue,))
    p0.start()
    p1 = Process(target=new, args=(queue,))
    p1.start()
    p2 = Process(target=new, args=(queue,))
    p2.start()

    p0.join()
    p1.join()
    p2.join()

    print("Done in --- %s seconds ---" % (time.time() - start_time))

保留print(shared_queue.qsize())语句运行时，能得到预期结果：

Consumer: Done
Consumer: Done
Consumer: Done
Done in --- 5.304457664489746 seconds ---

但仅注释该print语句后，运行结果变为：

Init Queue size 50000
Consumer: Running
Consumer: Running
Consumer: Running
Consumer: Done

为何无法看到其余的Consumer: Done语句以及最后的耗时统计信息？

核心问题出在**qsize()的非原子性判断**，以及循环条件的逻辑缺陷：

• qsize()不能作为可靠的循环终止条件
  multiprocessing.Queue.qsize()返回的队列大小是近似值，多进程环境下，队列状态可能在调用qsize()和后续get()操作之间发生变化，这个判断不是原子操作，存在竞争窗口。

• 有无print语句的差异本质

  • 保留print时，打印操作会引入短暂延迟，弱化了多进程间的竞争冲突，大概率让所有进程在队列空时退出循环。
  • 去掉print后，进程执行速度极快，容易出现这种场景：队列只剩最后一个元素时，多个进程同时通过while shared_queue.qsize() > 0的判断；其中一个进程执行get()把队列拿空，剩下的进程执行get()时会永久阻塞（Queue.get()默认是阻塞模式，队列空时会一直等待新元素），永远不会退出循环，自然不会打印Consumer: Done；主进程的join()会一直等待这些阻塞的进程，所以最后的耗时统计也无法输出。

• 正确的实现方式
  不要依赖qsize()判断循环结束，推荐两种方案：

  1. 给get()添加超时机制，超时则判定队列已空并退出：

  from random import random
  from multiprocessing import Process
  from multiprocessing import Queue
  import time
  
  def new(shared_queue):
      print('Consumer: Running', flush=True)
      while True:
          try:
              # 设置1秒超时，避免队列空时永久阻塞
              item = shared_queue.get(timeout=1)
          except:
              # 超时说明队列已空，退出循环
              break
      print('Consumer: Done', flush=True)
  
  if __name__ == '__main__':
      start_time = time.time()
  
      queue = Queue()
      for _ in range(50000):
          value = random()
          queue.put(value)
  
      print('Init Queue size', queue.qsize())
  
      p0 = Process(target=new, args=(queue,))
      p0.start()
      p1 = Process(target=new, args=(queue,))
      p1.start()
      p2 = Process(target=new, args=(queue,))
      p2.start()
  
      p0.join()
      p1.join()
      p2.join()
  
      print("Done in --- %s seconds ---" % (time.time() - start_time))
  

  2. 向队列中添加与进程数对应的终止标记（比如None），进程拿到标记即退出：

  from random import random
  from multiprocessing import Process
  from multiprocessing import Queue
  import time
  
  def new(shared_queue):
      print('Consumer: Running', flush=True)
      while True:
          item = shared_queue.get()
          if item is None:
              # 拿到终止标记，退出循环
              break
          # 处理拿到的元素
      print('Consumer: Done', flush=True)
  
  if __name__ == '__main__':
      start_time = time.time()
  
      queue = Queue()
      for _ in range(50000):
          value = random()
          queue.put(value)
  
      # 添加3个终止标记，对应3个消费者进程
      for _ in range(3):
          queue.put(None)
  
      print('Init Queue size', queue.qsize())
  
      p0 = Process(target=new, args=(queue,))
      p0.start()
      p1 = Process(target=new, args=(queue,))
      p1.start()
      p2 = Process(target=new, args=(queue,))
      p2.start()
  
      p0.join()
      p1.join()
      p2.join()
  
      print("Done in --- %s seconds ---" % (time.time() - start_time))
  

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