Python ThreadPoolExecutor中复用Whoosh Searcher对象的实现问题

问题根源 #

你遇到的UnpicklingError: could not find MARK确实是因为多线程共用了同一个Searcher对象。你的代码中，MyExecutor(whoosh_index)是在主线程初始化的，所有线程都会复用这个实例里的self.searcher——而Whoosh的Searcher包含文件指针这类线程不安全的内部状态，多线程同时操作会打乱文件读取位置，引发上述异常。

正确实现方式：线程本地存储复用Searcher #

要实现每个线程拥有独立Searcher且复用，可以用Python的threading.local()创建线程本地存储，每个线程会持有自己的Searcher实例，线程复用时直接复用已创建的实例。

修改后的代码如下：

from whoosh import Index
from concurrent.futures import ThreadPoolExecutor
from typing import List
import threading
from functools import partial

# 线程本地存储，每个线程独立拥有一份数据
_thread_local = threading.local()

def _get_thread_searcher(index: Index):
    # 检查当前线程是否已创建Searcher，没有则创建
    if not hasattr(_thread_local, "searcher"):
        _thread_local.searcher = index.searcher()
    return _thread_local.searcher

def _process_query(query, index: Index):
    searcher = _get_thread_searcher(index)
    return searcher.search(query)

def query_batched(queries: List[str], whoosh_index: Index, num_threads: int):
    with ThreadPoolExecutor(max_workers=num_threads) as pool:
        # 绑定索引参数，避免变量捕获问题
        process_func = partial(_process_query, index=whoosh_index)
        return list(pool.map(process_func, queries))

代码说明 #

• threading.local()：创建的对象会为每个线程维护独立的属性空间，确保每个线程的Searcher互不干扰。
• _get_thread_searcher：每个线程第一次调用时创建Searcher并存储，后续调用直接返回已有的实例，实现复用。
• partial：用来绑定whoosh_index参数，避免lambda表达式中变量捕获导致的潜在问题。

额外注意事项 #

• Whoosh的Searcher在被垃圾回收时会自动关闭文件，线程复用场景下无需手动关闭，否则会失去复用的性能优势。
• 确保你的Query对象是不可变的（Whoosh默认Query对象为不可变类型），可安全在多线程间传递。

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