如何在多线程生产者 - 使用者模式下完成工作线程后退出工作线程？

Question

我试图在Python 2.7中使用 Queue.Queue实现多线程生产者 - 消费者模式。我试图弄清楚如何让 使用者（即工作者线程）在完成所有必需的工作后停止。

见马丁詹姆斯第二评论这个答案：https://stackoverflow.com/a/19369877/1175080

发送“我完了”的任务，指示池中的线程终止。任何得到这样的任务的线程都会重新执行它，然后自杀。

但这不适用于我。例如，请参阅以下代码。

import Queue 
import threading 
import time 

def worker(n, q): 
    # n - Worker ID 
    # q - Queue from which to receive data 
    while True: 
     data = q.get() 
     print 'worker', n, 'got', data 
     time.sleep(1) # Simulate noticeable data processing time 
     q.task_done() 
     if data == -1: # -1 is used to indicate that the worker should stop 
      # Requeue the exit indicator. 
      q.put(-1) 
      # Commit suicide. 
      print 'worker', n, 'is exiting' 
      break 

def master(): 
    # master() sends data to worker() via q. 
    q = Queue.Queue() 

    # Create 3 workers. 
    for i in range(3): 
     t = threading.Thread(target=worker, args=(i, q)) 
     t.start() 

    # Send 10 items to work on. 
    for i in range(10): 
     q.put(i) 
     time.sleep(0.5) 

    # Send an exit indicator for all threads to consume. 
    q.put(-1) 

    print 'waiting for workers to finish ...' 
    q.join() 
    print 'done' 

master() 

该程序挂起所有三个工人已经阅读出口指示灯， 即-1从队列后，因为每个工人重新排队-1前 退出，所以队列永远不会成为空和q.join()永远不会返回。

我想出了以下但丑陋的解决方案，我通过队列发送-1出口 指标为每名工人，使每个工人可以看到它 和自杀。但是，我必须为每个工作人员发送退出指示 ，这一事实感觉有点难看。

import Queue 
import threading 
import time 

def worker(n, q): 
    # n - Worker ID 
    # q - Queue from which to receive data 
    while True: 
     data = q.get() 
     print 'worker', n, 'got', data 
     time.sleep(1) # Simulate noticeable data processing time 
     q.task_done() 
     if data == -1: # -1 is used to indicate that the worker should stop 
      print 'worker', n, 'is exiting' 
      break 

def master(): 
    # master() sends data to worker() via q. 
    q = Queue.Queue() 

    # Create 3 workers. 
    for i in range(3): 
     t = threading.Thread(target=worker, args=(i, q)) 
     t.start() 

    # Send 10 items to work on. 
    for i in range(10): 
     q.put(i) 
     time.sleep(0.5) 

    # Send one stop indicator for each worker. 
    for i in range(3): 
     q.put(-1) 

    print 'waiting for workers to finish ...' 
    q.join() 
    print 'done' 

master() 

我有两个问题。

1. 是否可以为所有线程发送单个退出指示符的方法（如Martin James的https://stackoverflow.com/a/19369877/1175080的第二条评论所述）甚至可以工作？
2. 如果上一个问题的答案是“否”，是否有办法解决问题，我不必为每个工作线程发送单独的退出指示符？

Answer 1

不要将它称为任务的特例。

改为使用Event，为您的员工实施非阻塞实施。

stopping = threading.Event() 

def worker(n, q, timeout=1): 
    # run until the master thread indicates we're done 
    while not stopping.is_set(): 
     try: 
      # don't block indefinitely so we can return to the top 
      # of the loop and check the stopping event 
      data = q.get(True, timeout) 
     # raised by q.get if we reach the timeout on an empty queue 
     except queue.Empty: 
      continue 
     q.task_done() 

def master(): 
    ... 

    print 'waiting for workers to finish' 
    q.join() 
    stopping.set() 
    print 'done' 

Answer 2

可以发送一个出口指标的所有线程（如马丁詹姆斯https://stackoverflow.com/a/19369877/1175080第二注释解释）的方法，甚至工作？

正如你已经注意到它不能工作，传播消息将使最后一个线程更新队列与一个更多的项目，因为你正在等待一个永远不会是空的队列，而不是你的代码有。

如果上一个问题的答案是“否”，是否有办法解决问题的方式，我不必为每个工作线程发送单独的退出指示符？

可以join线程，而不是队列：

def worker(n, q): 
    # n - Worker ID 
    # q - Queue from which to receive data 
    while True: 
     data = q.get() 
     print 'worker', n, 'got', data 
     time.sleep(1) # Simulate noticeable data processing time 
     q.task_done() 
     if data == -1: # -1 is used to indicate that the worker should stop 
      # Requeue the exit indicator. 
      q.put(-1) 
      # Commit suicide. 
      print 'worker', n, 'is exiting' 
      break 

def master(): 
    # master() sends data to worker() via q. 
    q = Queue.Queue() 

    # Create 3 workers. 
    threads = [threading.Thread(target=worker, args=(i, q)) for i in range(3)] 
    for t in threads: 
     threads.start() 
    # Send 10 items to work on. 
    for i in range(10): 
     q.put(i) 
     time.sleep(0.5) 

    # Send an exit indicator for all threads to consume. 
    q.put(-1) 

    print 'waiting for workers to finish ...' 
    for t in threads: 
     t.join() 
    print 'done' 

master() 

由于Queue documentation解释get方法将上升，一旦其空的execption所以如果你已经知道的数据处理您可以填写队列和那么垃圾邮件的主题：

import Queue 
import threading 
import time 

def worker(n, q): 
    # n - Worker ID 
    # q - Queue from which to receive data 
    while True: 
     try: 
      data = q.get(block=False, timeout=1) 
      print 'worker', n, 'got', data 
      time.sleep(1) # Simulate noticeable data processing time 
      q.task_done() 
     except Queue.Empty: 
      break 


def master(): 
    # master() sends data to worker() via q. 
    q = Queue.Queue() 

    # Send 10 items to work on. 
    for i in range(10): 
     q.put(i) 

    # Create 3 workers. 
    for i in range(3): 
     t = threading.Thread(target=worker, args=(i, q)) 
     t.start() 

    print 'waiting for workers to finish ...' 
    q.join() 
    print 'done' 

master() 

这里有一个live example

Answer 3

除了@DanielSanchez出色答卷，我建议实际上 依靠类似的机制为Java CountDownLatch。

要点之中，

• 创建latch将只开放以后一定计数器下去，

• 当锁被打开时，该线程（或多个）等待它将被允许继续执行。

• 我做了一个过于简单的例子，检查here一类像例子这样的锁存：

  import threading 
  import Queue 
  import time 
  
  WORKER_COUNT = 3 
  latch = threading.Condition() 
  count = 3 
  
  def wait(): 
      latch.acquire() 
      while count > 0: 
       latch.wait() 
      latch.release() 
  
  def count_down(): 
      global count 
      latch.acquire() 
      count -= 1 
      if count <= 0: 
       latch.notify_all() 
      latch.release() 
  
  def worker(n, q): 
      # n - Worker ID 
      # q - Queue from which to receive data 
      while True: 
       data = q.get() 
       print 'worker', n, 'got', data 
       time.sleep(1) # Simulate noticeable data processing time 
       q.task_done() 
       if data == -1: # -1 is used to indicate that the worker should stop 
        # Requeue the exit indicator. 
        q.put(-1) 
        # Commit suicide. 
        count_down() 
        print 'worker', n, 'is exiting' 
        break 
  
  # master() sends data to worker() via q. 
  
  def master(): 
      q = Queue.Queue() 
  
      # Create 3 workers. 
      for i in range(WORKER_COUNT): 
       t = threading.Thread(target=worker, args=(i, q)) 
       t.start() 
  
      # Send 10 items to work on. 
      for i in range(10): 
       q.put(i) 
       time.sleep(0.5) 
  
      # Send an exit indicator for all threads to consume. 
      q.put(-1) 
      wait() 
      print 'done' 
  
  master() 
  

Answer 4

只是为了完整性缘故 你也可以排队停止信号是 - （线程数）。 每个线程则可以通过一个递增并重新排队它只有在停止信号！= 0

if data < 0: # negative numbers are used to indicate that the worker should stop 
     if data < -1: 
      q.put(data + 1) 
     # Commit suicide. 
     print 'worker', n, 'is exiting' 
     break 

但我会亲自去与Travis Mehlinger或 Daniel Sanchez答案。