为什么这里需要Queue.join()？

Question

我正在学习python的线程模块，并编写了以下代码来帮助自己理解

from Queue import Queue
import threading

lock = threading.Lock()
MAX_THREADS = 8
q = Queue()
count = 0

# some i/o process
def io_process(x):
    pass

# process that deals with shared resources
def shared_resource_process(x):
    pass

def func():
    global q, count
    while not q.empty():
        x = q.get()
        io_process(x)
        if lock.acquire():
            shared_resource_process(x)
            print '%s is processing %r' %(threading.currentThread().getName(), x)
            count += 1          
            lock.release()

def main():
    global q
    for i in range(40):
        q.put(i)

    threads = []
    for i in range(MAX_THREADS):
        threads.append(threading.Thread(target=func))

    for t in threads:
        t.start()

    for t in threads:
        t.join()

    print 'multi-thread done.'
    print count == 40

if __name__ == '__main__':
    main()

输出就像这样卡住了：

Thread-1 is processing 32
Thread-8 is processing 33
Thread-6 is processing 34
Thread-2 is processing 35
Thread-5 is processing 36
Thread-3 is processing 37
Thread-7 is processing 38
Thread-4 is processing 39

注意，main()中的打印没有执行，这意味着有些线程挂起了/blocking？

然后通过添加q.task_done()来修改func()方法：

if lock.acquire():
            shared_resource_process(x)
            print '%s is processing %r' %(threading.currentThread().getName(), x)
            count += 1
            q.task_done()  # why is this necessary ?
            lock.release()

现在，所有线程都如我所期望的那样终止，并得到正确的输出：

Thread-6 is processing 36
Thread-4 is processing 37
Thread-3 is processing 38
Thread-7 is processing 39
multi-thread done.
True

Process finished with exit code 0

我阅读了Queue.Queue 这里的文档，并看到task_done()与queue.join()一起工作，以确保队列中的所有项都被处理。但是，由于我没有在main()中调用queue.join()，为什么task_done()在func()中是必需的？当我错过task_done()代码时，线程挂起/阻塞的原因是什么？

Answer 1

您的代码中有一个种族条件。假设Queue中只剩下一个项，并且只使用两个线程而不是8个线程。

1. 线程A调用q.empty来检查它是否为空。由于队列中有一个项，结果是False，因此执行循环体。
2. 在线程A调用q.get之前，有一个上下文开关，线程B可以运行。
3. 线程B调用q.empty，队列中仍然有一个项，因此结果是False并执行循环体。
4. 线程B不带参数地调用q.get，它将立即返回队列中的最后一个项。然后线程B处理项目并退出，因为q.empty返回True。
5. 线程A可以运行。因为它已经在步骤1中调用了q.empty，所以它将调用q.get下一步，但是这将永远阻塞，因此您的程序不会终止。

您可以通过导入time并稍微更改循环来模拟上述行为：

while not q.empty():
    time.sleep(0.1) # Force context switch
    x = q.get()

请注意，无论是否调用task_done，行为都是相同的。

那么为什么添加task_done会有所帮助呢？默认情况下，Python 2将每100个解释器指令进行上下文切换，因此添加代码可能会改变发生上下文切换的位置。有关更好的解释，请参见另一个问题和链接PDF。在我的机器上，程序没有挂起，不管task_done是否在那里，所以这只是一个猜测，是什么导致了它的发生。

如果您想要修复该行为，您可以只使用无限循环并将参数传递给get，指示它不要阻塞。这导致get最终抛出可以捕获的Queue.Empty异常，然后中断循环：

from Queue import Queue, Empty

def func():
    global q, count
    while True:
        try:
            x = q.get(False)
        except Empty:
            break
        io_process(x)
        if lock.acquire():
            shared_resource_process(x)
            print '%s is processing %r' %(threading.currentThread().getName(), x)
            count += 1
            lock.release()