Python空队列q.get()无返回导致主线程在queue join处挂起问题

问题核心原因 #

你遇到的两个核心问题根源如下：

1. q.get() 不返回的原因：Python标准库queue.Queue的get()方法默认参数为block=True, timeout=None，当队列为空时，调用该方法会永久阻塞等待新元素入队，永远不会抛出queue.Empty异常。只有显式设置block=False或者指定timeout参数且超时后，才会抛出queue.Empty异常，这就是你空队列下get()卡住的直接原因。
2. q.join()挂起的原因：
   • 你在清空队列剩余元素时，只调用了1次q.task_done()，但队列的unfinished_tasks计数是每put一个元素就加1的，假设清空时队列还有N个未处理元素，就需要调用N次task_done()才能让计数归零，否则q.join()会一直等待计数归零，自然会挂起。
   • 其他工作线程可能在你清空队列前就已经调用了阻塞的q.get()，就算队列被清空，这些线程也会一直陷入阻塞等待，永远不会退出，进一步导致join()无法返回。

修复方案 #

推荐用「非阻塞get+终止事件」的方案实现安全的提前退出，避免竞态条件和阻塞问题，修改后的核心代码如下：

1. 修改RequestManager逻辑 #

import threading
import queue

class RequestManager:
    def __init__(self, config=None):
        self.config = config

    def request_all_heroes(self):
        q = queue.Queue()
        result_list = []
        # 新增终止事件，用于通知所有线程提前退出
        stop_event = threading.Event()
        
        for skip in [x * 100 for x in range(1, 3)]:
            q.put_nowait(skip)

        thread_num = int(self.config["meta"]["number_of_threads"])
        for _ in range(thread_num):
            RequestWorker(
                q=q,
                config=self.config,
                query_name='all_heroes',
                shared_result_list=result_list,
                stop_event=stop_event,
                daemon=True # 设为守护线程，避免极端情况进程卡死
            ).start()
        q.join()
        return [Hero(item) for sublist in result_list for item in sublist]

2. 修改RequestWorker逻辑 #

class RequestWorker(threading.Thread):
    def __init__(self,
                 q=None,
                 config=None,
                 query_name="",
                 shared_result_list=None,
                 stop_event=None,
                 *args, **kwargs):
        self.q = q
        self.config = config
        self.query_file_path = self.config["files"][query_name]
        self.shared_result_list = shared_result_list
        self.stop_event = stop_event
        super().__init__(*args, **kwargs)

    def run(self):
        # 循环先检查终止事件，触发就直接退出
        while not self.stop_event.is_set():
            try:
                # 非阻塞get，队列为空直接抛异常退出
                skip_number = self.q.get(block=False)
            except queue.Empty:
                return

            sr = SpecificRequest(config=self.config, skip=skip_number, query_file_path=self.query_file_path)
            request_result = sr.do_specific_request()

            if len(request_result) == 0:
                # 触发终止事件，通知所有线程退出
                self.stop_event.set()
                with self.q.mutex:
                    # 统计剩余未处理的任务数
                    remaining_tasks = len(self.q.queue)
                    self.q.queue.clear()
                    # 把剩余任务的task_done全部补上，让join能正常返回
                    for _ in range(remaining_tasks):
                        self.q.task_done()
                # 标记当前任务完成
                self.q.task_done()
                return

            self.shared_result_list.append(request_result)
            self.q.task_done()

方案说明 #

• 非阻塞的get(block=False)保证队列空时线程会直接抛出异常退出，不会陷入永久阻塞。
• 终止事件stop_event是线程安全的通知机制，不会出现竞态条件，所有线程每次循环都会先检查是否要终止。
• 清空队列时统计剩余任务数，批量调用对应次数的task_done()，保证unfinished_tasks计数能正确归零，q.join()可以正常返回。

内容的提问来源于stack exchange，提问作者Lennart