使用多进程操作MongoDB时遇TypeError: can't pickle _thread.lock objects错误

我之前处理MongoDB多进程任务时也踩过这个坑，这个错误的核心原因很明确：MongoDB的客户端（比如pymongo.MongoClient）内部包含线程锁这类无法被Python序列化（pickle）的对象。当你用多进程时，父进程的连接会被尝试传递给子进程，而pickle机制搞不定这类锁对象，直接就抛出这个错误了。

下面给你几个可行的修复方案，按推荐程度排序：

1. 每个子进程单独创建MongoDB连接 #

这是最稳妥的方案——别在父进程里建好连接再传给子进程，让每个子进程自己初始化连接。这样完全避开了序列化连接对象的问题。

修改你的代码逻辑大概是这样：

from multiprocessing import Pool
import pymongo

def save_for_doc(doc_id):
    # 子进程内部单独初始化MongoDB连接
    client = pymongo.MongoClient("mongodb://localhost:27017/")
    db = client["你的数据库名"]
    
    # 获取文档字段（确保get_fields里也没用到父进程的连接）
    fields = get_fields(doc_id)
    
    # 这里写你的值比较、计算逻辑
    # ... 比如对比字段值、生成计算结果 ...
    
    # 将结果存入目标集合
    result_col = db["结果集合名"]
    result_col.insert_one({"doc_id": doc_id, "fields": fields, "计算结果": ...})
    
    # 可选：关闭连接，不过pymongo会自动管理连接池，不手动关也没问题
    client.close()

if __name__ == "__main__":
    no_of_process = 5
    doc_col_size = 30000
    chunk_size = round(doc_col_size / no_of_process)
    
    # 替换成你实际的doc_id列表
    doc_ids = [i for i in range(doc_col_size)]
    
    with Pool(no_of_process) as pool:
        pool.map(save_for_doc, doc_ids)

2. 用进程池初始化函数统一创建连接 #

如果不想在每个函数里重复写连接代码，可以用Pool的initializer参数，让每个子进程启动时自动初始化连接：

from multiprocessing import Pool
import pymongo

# 全局变量，每个子进程会有自己的这个变量实例
client = None

def init_worker():
    """每个子进程启动时执行的初始化函数"""
    global client
    client = pymongo.MongoClient("mongodb://localhost:27017/")

def save_for_doc(doc_id):
    db = client["你的数据库名"]
    fields = get_fields(doc_id)
    
    # ... 你的业务逻辑 ...
    
    result_col = db["结果集合名"]
    result_col.insert_one(你的结果字典)

if __name__ == "__main__":
    no_of_process = 5
    doc_col_size = 30000
    doc_ids = [i for i in range(doc_col_size)]
    
    # 初始化进程池时指定初始化函数
    with Pool(no_of_process, initializer=init_worker) as pool:
        pool.map(save_for_doc, doc_ids)

3. 检查get_fields函数的依赖 #

还要注意你的get_fields函数，如果它依赖父进程的MongoDB连接，或者返回了MongoDB游标、文档对象这类不可序列化的内容，也会触发同样的错误。确保get_fields要么在子进程内执行（用子进程自己的连接），要么只返回普通的可序列化数据（比如字典、字符串、数字）。

几个关键提醒 #

• MongoDB的客户端连接是线程安全但进程不安全的，绝对不能在多进程之间共享同一个连接实例。
• pymongo默认自带连接池，所以每个子进程创建连接的开销其实不大，不用太担心性能问题。
• 如果处理的文档数量很多，建议批量处理，减少频繁插入的开销，比如攒一批结果再一次性插入。

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