确定线程池何时完成对queu的处理

Question

我正在尝试实现一个线程池，该线程池使用ThreadPool和Queue处理任务队列。它从一个初始任务队列开始，然后每个任务也可以将其他任务推送到任务队列中。问题是，在队列为空且线程池已完成处理之前，我不知道如何阻止，但仍然会检查队列并将任何新任务提交到推送到队列上的线程池。我不能简单地调用ThreadPool.join()，因为我需要为新任务打开池。

例如：

from multiprocessing.pool import ThreadPool
from Queue import Queue
from random import random
import time
import threading

queue = Queue()
pool = ThreadPool()
stdout_lock = threading.Lock()

def foobar_task():
    with stdout_lock: print "task called" 
    if random() > .25:
        with stdout_lock: print "task appended to queue"
        queue.append(foobar_task)
    time.sleep(1)

# set up initial queue
for n in range(5):
    queue.put(foobar_task)

# run the thread pool
while not queue.empty():
    task = queue.get() 
    pool.apply_async(task)

with stdout_lock: print "pool is closed"
pool.close()
pool.join()

这将输出：

pool is closed
task called
task appended to queue
task called
task appended to queue
task called
task appended to queue
task called
task appended to queue
task called
task appended to queue

这将在foobar_任务附加到队列之前退出while循环，因此附加的任务永远不会提交到线程池。我找不到任何方法来确定线程池是否仍有任何活动的工作线程。我尝试了以下方法：

while not queue.empty() or any(worker.is_alive() for worker in pool._pool):
    if not queue.empty():
        task = queue.get() 
        pool.apply_async(task)
    else:   
        with stdout_lock: print "waiting for worker threads to complete..."
        time.sleep(1)

但似乎worker.is_alive()总是返回true，所以这进入了一个无限循环。

有更好的办法吗？

Answer 1

1. 在处理完每个任务后调用queue.task_done。
2. 然后可以调用queue.join()来阻塞主线程，直到 任务已完成。
3. 要终止工作线程，请在队列中放置一个sentinel（例如None）， 当它接收到哨兵时，有foobar_task从while-loop中断。
4. 我认为用threading.Threads比用ThreadPool更容易实现。

---

import random
import time
import threading
import logging
import Queue

logger=logging.getLogger(__name__)
logging.basicConfig(level=logging.DEBUG)

sentinel=None
queue = Queue.Queue()
num_threads = 5

def foobar_task(queue):
    while True:
        n = queue.get()
        logger.info('task called: {n}'.format(n=n))
        if n is sentinel: break
        n=random.random()
        if n > .25:
            logger.info("task appended to queue")
            queue.put(n)
        queue.task_done()

# set up initial queue
for i in range(num_threads):
    queue.put(i)

threads=[threading.Thread(target=foobar_task,args=(queue,))
         for n in range(num_threads)]
for t in threads:
    t.start()

queue.join()
for i in range(num_threads):
    queue.put(sentinel)

for t in threads:
    t.join()
logger.info("threads are closed")