Python多线程无法正确从Queue取任务 重复处理PDF求优化方案

• queue.task_done() 调用位置错误：只有成功处理完一个队列元素后才需要调用task_done()，原代码仅在队列为空时调用，完全不符合Queue的使用规范，会导致队列状态异常。
• 递归消费逻辑存在隐患：原代码在searchName里用递归判断队列非空就调用自身，配合35个高并发线程，会出现多个线程同时读取队列状态、打印qsize()时序重叠的问题。qsize()本身在多线程并发场景下仅返回近似值，多次打印重复数值属于时序重叠，并非元素被重复处理。
• 跨线程操作UI风险：所有listWidget.addItem操作都在子线程直接调用，Qt的UI组件仅允许在主线程操作，轻则UI刷新异常，重则程序直接崩溃。
• 预序列化对象不合理：PyPDF2的PdfFileReader对象绑定了文件句柄和运行时状态，序列化到本地再反序列化极易出现不可预期的IO错误，反而比运行时初始化Reader对象效率更低、稳定性更差。
• 线程数设置不合理：PDF文本提取是CPU密集型任务，线程数设置为CPU核心数的1~2倍即可，35个线程会导致大量上下文切换开销，反而拉低整体处理效率。

修复后核心代码 #

import threading
import queue
import os
import re
import pickle
import PyPDF2
from queue import Empty

# 线程锁保护UI操作
ui_lock = threading.Lock()

def searchButtonClicked(self):
    name = self.lineEdit.text()
    self.listWidget.addItem(f"Searching with the name: {name}")
    # CPU密集型任务线程数匹配CPU核心数即可
    num_threads = min(16, os.cpu_count() or 4)
    self.search_stop = False # 新增停止标识，支持中途终止任务
    
    # 队列初始化逻辑优化
    try:
        with open("my_saved_queue.obj","rb") as queue_save_file:
            # 缓存仅存文件路径，不存PyPDF2对象
            pdf_paths: list = pickle.load(queue_save_file)
            self.my_loaded_queue = queue.Queue()
            for path in pdf_paths:
                self.my_loaded_queue.put(path)
    except:
        pdf_paths = self.saveFile()
        self.my_loaded_queue = queue.Queue()
        for path in pdf_paths:
            self.my_loaded_queue.put(path)
    
    # 启动工作线程
    self.workers = []
    for i in range(num_threads):
        worker = threading.Thread(target=self.searchName, args=(name,), daemon=True)
        worker.start()
        self.workers.append(worker)
    
    # 主线程等待队列处理完成，统一提示结束
    def wait_finish():
        self.my_loaded_queue.join()
        with ui_lock:
            self.listWidget.addItem("---------------Done---------------")
    threading.Thread(target=wait_finish, daemon=True).start()

def saveFile(self):
    pdf_paths = []
    for root, dirs, files in os.walk(self.directory):
        for file_name in files:
            file_path = os.path.join(root, file_name)
            if file_path.endswith(".pdf"):
                pdf_paths.append(file_path)
    with open("my_saved_queue.obj","wb+") as queue_save_file:
        pickle.dump(pdf_paths, queue_save_file)
    return pdf_paths

def searchName(self, name):
    # 用循环替代递归，避免栈溢出
    while not self.search_stop:
        try:
            file_path = self.my_loaded_queue.get(timeout=1)
        except Empty:
            break
        try:
            # 运行时初始化PdfFileReader，稳定性更高
            with open(file_path, 'rb') as f:
                pdf_object = PyPDF2.PdfFileReader(f, strict=False)
                num_of_pages = pdf_object.getNumPages()
                for i in range(num_of_pages):
                    page_obj = pdf_object.getPage(i)
                    text = page_obj.extractText() 
                    if re.search(name, text):
                        res = f"{file_path}  Page {str(i+1)}"
                        print(res)
                        with ui_lock:
                            self.listWidget.addItem(res)
                        # 匹配到关键词后直接跳出当前PDF遍历，大幅提升检索效率
                        break
        except Exception as e:
            print(f"处理文件{file_path}失败：{str(e)}")
        finally:
            # 无论处理成功失败都调用task_done，保证队列状态正确
            self.my_loaded_queue.task_done()
            with ui_lock:
                print(self.my_loaded_queue.qsize())

def clearListWidget(self):
    self.search_stop = True # 清空列表前先终止所有工作线程
    for worker in self.workers:
        worker.join(timeout=1)
    self.listWidget.clear()

核心优化点 #

• 修复队列状态管理逻辑：把task_done()移到每个元素处理完成后的finally块中，符合queue.Queue的使用规范，队列join()可以准确感知所有元素处理完成。
• 替换递归为循环消费：避免递归深度过大导致栈溢出，逻辑更清晰，彻底避免重复消费风险。
• 优化序列化存储内容：仅存PDF文件路径，运行时再初始化PdfFileReader，解决序列化对象带来的稳定性问题，整体处理效率反而更高。
• 线程数合理化配置：根据CPU核心数设置线程数，减少不必要的上下文切换开销。
• 加UI操作锁：所有跨线程操作Qt UI的逻辑都加锁保护，避免UI崩溃。
• 增加提前终止逻辑：匹配到关键词后直接跳出当前PDF的页面遍历，不用扫描全文档，大幅提升检索效率。
• 统一处理完成提示：由单独的等待线程统一触发完成提示，避免多个子线程重复打印完成信息。

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