为什么这里需要使用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()函数中的打印语句未被执行，这意味着一些线程可能正在挂起/阻塞？

然后我通过添加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() 一起使用可以确保队列中的所有项都被处理。但是，既然我没有在主函数中调用 queue.join()，为什么在 func() 中需要 task_done() 呢？如果我忘记了 task_done() 代码会导致线程挂起或阻塞的原因是什么？

- vansdev

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()

- niemmi