多进程队列+线程实现网页下载时进程/线程挂起问题排查

核心挂起原因 #

1. SENTINEL数量不足，剩余线程阻塞在队列get()
   你启动了10个工作线程，但只向to_download队列放入了1个SENTINEL。只有第一个拿到SENTINEL的线程会退出循环，剩下的9个线程会一直阻塞在self.to_download.get()调用上（队列已空且无新元素），导致download_items中的thread.join()永远无法完成，进程卡在这一步，无法执行后续打印和向downloaded队列放入SENTINEL的逻辑。

2. _stop_workers无法唤醒已阻塞的线程
   即便第一个线程设置了self._stop_workers = True，其他线程此时已经进入to_download.get()的阻塞状态——这个阻塞调用不会主动检查_stop_workers的值，直到队列有新元素放入，所以这些线程不会因_stop_workers变为True而退出阻塞。

3. 跨进程队列的多线程消费缺陷
   你使用的multiprocessing.Queue本身线程安全，但多线程消费场景下，必须保证每个工作线程都能收到终止信号（SENTINEL），否则未收到信号的线程会一直阻塞在get操作。

修复方案 #

• 为每个工作线程匹配一个SENTINEL：生产者进程结束添加任务后，向to_download队列放入与工作线程数量相同的SENTINEL（比如10个），确保每个线程都能拿到终止信号并退出循环。
• 修改线程循环逻辑，避免阻塞时无法响应终止：使用Queue.get(timeout)配合循环检查_stop_workers，让线程定期从阻塞中醒来检查终止条件。

示例修改后的核心逻辑：

# 生产者进程修改：添加完下载项后放入对应数量的SENTINEL
def add_item_to_download():
    # ... 原有添加下载项的逻辑 ...
    # 对应10个工作线程，放入10个终止信号
    for _ in range(10):
        to_download.put(END_QUEUE_SENTINEL)

# Downloader类的_download_item方法修改
def _download_item(self):
    while not self._stop_workers:
        try:
            # 设置超时，定期检查_stop_workers
            item = self.to_download.get(timeout=1)
            if item == self.SENTINEL:
                print("sentinel received")
                # 用with语句更安全地管理锁
                with self.mutex:
                    self._stop_workers = True
                print("self._stop_workers becomes True")
                print(self.to_download.qsize())
                break
            req = urlrequest.Request(item.url)
            response = urlrequest.urlopen(req)
            item = self._set_item(response, item)
            self.downloaded.put(item)
        except queue.Empty:
            # 超时后继续循环检查终止条件
            continue

额外注意点 #

• 代码中item: = self.to_download.get()存在语法错误（多余冒号），需修正为item = self.to_download.get()。
• 使用with self.mutex:管理锁比手动acquire/release更安全，能避免异常导致锁无法释放的问题。

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