Python multiprocessing结合MongoDB并行处理文件的正确方法

当前代码重复处理文件的核心问题有三个：

• 竞态条件导致重复抢占：你现在的逻辑是先查询文件状态、判断为new后再更新为processing，这是三个独立的非原子操作。多个进程可能在同一个时间窗口内查到同一个文件的状态都是new，随后都会执行更新和处理逻辑，直接导致重复拾取。
• 任务分发逻辑冗余：所有进程都独立遍历目标路径下的全量jpg文件，既重复消耗磁盘IO，也进一步放大了竞态问题的出现概率。
• 跨进程Mongo连接隐患：你在父进程中初始化MongoClient后直接在子进程中复用，Mongo连接实例不支持跨进程安全调用，运行过程中可能出现连接失效、请求错乱的隐性bug。

1. 用数据库原子操作抢占任务，从根源避免重复 #

不要分开做查询和更新，把「判断状态为new+更新为processing」合并为单个Mongo原子操作，用find_one_and_update实现：只有当文件状态确实是new时才会更新为processing，同时返回对应文档；如果更新结果为空，说明文件已经被其他进程抢走，直接跳过即可。
核心代码示例：

from pymongo import ReturnDocument

# 原子抢占任务
task = mycol.find_one_and_update(
    filter={"name": file, "status": "new"},
    update={"$set": {"status": "processing"}},
    return_document=ReturnDocument.AFTER
)
# 没抢到任务直接跳过
if not task:
    continue

2. 优化任务分发逻辑，避免全量重复遍历 #

不要让每个worker进程都扫一遍目录，推荐用主进程统一收集所有待处理文件路径，放入进程安全的任务队列，worker进程只需要从队列取任务处理即可，既减少重复IO，也方便做任务流控。

3. 修正Mongo连接初始化位置 #

MongoClient必须在每个子进程内部单独初始化，不能在父进程创建后跨进程复用，把连接初始化逻辑放到worker函数的开头即可。

import glob
import time
import multiprocessing as mp
import pymongo
from pymongo import ReturnDocument

# 替换为你的实际文件路径
PATH_TO_FILES = "/your/target/jpg/path"
# 替换为你的实际Mongo连接串
MONGO_CONN = "mongodb://localhost:27017/"

def worker(task_queue):
    # 每个子进程内部单独初始化Mongo连接
    myclient = pymongo.MongoClient(MONGO_CONN)
    mycol = myclient["file_list"]["file_list"]

    while True:
        # 从队列取任务，10秒没取到新任务就退出进程
        try:
            file_path = task_queue.get(timeout=10)
        except Exception:
            break

        # 原子抢占任务
        task = mycol.find_one_and_update(
            filter={"name": file_path, "status": "new"},
            update={"$set": {"status": "processing"}},
            return_document=ReturnDocument.AFTER
        )
        if not task:
            print(f"skip {file_path}: already picked by other process")
            continue

        print(f"start processing: {file_path}")
        # 此处插入你的文件处理逻辑
        # ...

        # 处理完成更新状态
        mycol.update_one(
            {"name": file_path},
            {"$set": {"status": "processed"}}
        )
        print(f"finish processing: {file_path}")

if __name__ == '__main__':
    start = time.time()
    num_processes = mp.cpu_count()
    task_queue = mp.Queue(maxsize=100) # 限制队列长度避免内存占用过高

    # 主进程扫描所有jpg文件放入队列
    all_jpg_files = glob.glob(f"{PATH_TO_FILES}/*.jpg")
    for jpg in all_jpg_files:
        task_queue.put(jpg)

    # 启动worker进程
    processes = []
    for _ in range(num_processes):
        p = mp.Process(target=worker, args=(task_queue,))
        processes.append(p)
        p.start()

    # 等待所有进程执行完成
    for p in processes:
        p.join()

    print(f"total time cost: {time.time() - start}s")

• 给Mongo集合的status字段加普通索引，可以大幅提升任务查询、更新的速度：mycol.create_index("status")
• 文件处理逻辑中增加异常捕获，如果处理过程中报错，把文件状态改回new或者标记为failed，避免任务一直卡在processing状态无法被重试。
• 如果待处理文件量级达到百万级以上，可以增加一个独立的生产者进程分批扫描目录、往队列放任务，不用一次性把所有文件路径加载到内存。

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