正确关闭守护线程

Question

我创建了自己的线程类，将在线程池中使用如下：

class SendFilesThread(threading.Thread):
    execute = True

    def __init__(self, port, rate, fileQueue):
        self.daemon = True
        self.fileQueue = fileQueue
        #Other initialization

    def run(self):
        while SendFilesThread.execute == True or not self.fileQueue.empty():
            self.filename = self.fileQueue.get()
            #some file processing
            self.fileQueue.task_done()

    @staticmethod
    def terminate():
        SendFilesThread.execute = False

在程序的主线程中，处理完所有文件后，我尝试关闭线程池，如下所示：

SendFilesThread.terminate()
for t in threadPool:
    if t.is_alive():
        t.join()
exit()

我的理解是，如果我调用join()，它将阻塞调用线程，在这种情况下是主线程，直到加入的线程完成处理。我的问题是，尽管线程完成了，但它永远不会返回到主线程并且程序只是挂起。关闭线程池时我做错了什么吗？

Answer 1

有两件事。首先，从设计的角度来看，让execute 标志位于类级别有点奇怪。通常最好将它作为每个类实例上的标志。

其次，你有一个竞争条件。 execute 标志由多个线程访问，并且不受同步原语（例如互斥锁）的保护。这意味着terminate() 调用可以在任何工作线程检查该标志之后运行（并设置标志），但在该线程尝试将下一个文件名出列。因为你在没有超时的情况下调用get()，所以工作线程会挂在这里，主线程会阻塞在t.join()调用中。出现死锁。

有很多方法可以解决这个问题。您可以使用线程同步原语（例如互斥锁）来保护execute 标志，或者使用threading.Event 对象代替简单的布尔值。

另一个，在我看来更简单的解决方案是在同一个队列上发送一个“哨兵”值，这表明线程应该退出。看起来您正在发送字符串文件名，因此空字符串可能是一个不错的选择。 （None也常用。）

每个线程的工作循环现在如下所示：

def run(self):
    while True:
        self.filename = self.fileQueue.get()
        if self.filename == '':
            return
        # Process file

与terminate() 静态方法不同，主线程为您拥有的每个工作线程放置一个空字符串（即fileQueue.enqueue('')）。

Answer 2

如果fileQueue 是multiprocessing.Queue 或queue.Queue，那么它的.get() 方法将阻塞，直到队列中有东西到达。尝试使用.get_nowait()，但如果您希望在队列为空的情况下继续执行，则可能必须将其包装在 try 块中。

try:
    self.filename = self.fileQueue.get_nowait()
except queue.Empty:
    time.sleep(1)

更多信息请参见queue docs