如何在Python中从进程返回字典？多进程倒排索引合并需求

你之前的问题核心是子进程生成的倒排索引字典无法传递回主进程，且共享字典方案性能不足。以下是两种高效实现分进程生成+主进程合并的方案：

方法一：使用multiprocessing.Pool自动收集结果 #

Pool的map方法会自动分发任务并收集每个子进程的返回值，代码简洁易维护，适合均匀分配任务的场景。

修改后的完整代码：

import multiprocessing as mp
from pathlib import Path
import re
import time
from collections import defaultdict


class InvertedIndex:
    def __init__(self):
        self.index = dict()

    def createIndex(self, path='data', threads_num=4):
        pathList = list(Path(path).glob('**/*.txt')) 
        fileNum = len(pathList)
        # 向上取整计算分片大小，避免最后一组无任务
        chunk_size = (fileNum + threads_num - 1) // threads_num
        # 分割文件列表为多个子任务组
        task_chunks = [pathList[i*chunk_size : min((i+1)*chunk_size, fileNum)] for i in range(threads_num)]
        
        # 创建进程池并执行任务
        with mp.Pool(processes=threads_num) as pool:
            # 收集所有子进程返回的索引字典
            sub_indices = pool.map(self.oneProcessTask, task_chunks)
        
        # 合并所有子索引到主索引
        for sub_idx in sub_indices:
            for word, doc_set in sub_idx.items():
                if word not in self.index:
                    self.index[word] = set()
                self.index[word].update(doc_set)

    @staticmethod
    def oneProcessTask(listOfDoc):
        # 使用defaultdict简化集合初始化逻辑
        tempDict = defaultdict(set)
        for name in listOfDoc:
            with open(name) as f:
                text = f.read()
                # 提取所有单词
                words = re.findall(r'\b\w+\b', text.lower())  # 统一小写避免大小写差异
                for w in words:
                    tempDict[w].add(str(name))
        # 将defaultdict转为普通字典返回
        return dict(tempDict)

    def getListOfDoc(self, keyWord):
        # 键不存在时返回空集合，避免KeyError
        return self.index.get(keyWord.lower(), set())


if __name__ == '__main__':
    ii = InvertedIndex()
    start_time = time.time()
    ii.createIndex()
    print("--- %s seconds ---" % (time.time() - start_time))
    # 测试示例
    # print(ii.getListOfDoc("test"))

关键改动说明： #

1. 替换手动创建Process的逻辑为Pool，利用map自动分发任务并收集返回值
2. 优化文件分片逻辑，确保每个进程任务量均匀
3. 让oneProcessTask返回生成的子索引字典，供主进程收集
4. 添加合并逻辑：遍历所有子字典，将每个词的文档集合合并到主索引
5. 使用collections.defaultdict简化集合初始化，减少冗余判断
6. 统一单词为小写，避免大小写导致的索引重复

方法二：使用multiprocessing.Queue手动传递结果 #

如果需要更灵活的进程控制（比如非均匀分配任务），可以用Queue手动传递子进程结果：

修改后的完整代码：

import multiprocessing as mp
from pathlib import Path
import re
import time
from collections import defaultdict


class InvertedIndex:
    def __init__(self):
        self.index = dict()

    def createIndex(self, path='data', threads_num=4):
        pathList = list(Path(path).glob('**/*.txt')) 
        fileNum = len(pathList)
        chunk_size = (fileNum + threads_num - 1) // threads_num

        processes = []
        result_queue = mp.Queue()  # 创建结果队列

        for i in range(threads_num):
            start = i * chunk_size
            end = min((i+1)*chunk_size, fileNum)
            curr_task = pathList[start:end]
            # 将队列作为参数传入子进程
            p = mp.Process(target=self.oneProcessTask, args=(curr_task, result_queue)) 
            processes.append(p)

        # 启动并等待所有进程完成
        [p.start() for p in processes]
        [p.join() for p in processes]

        # 从队列取出所有子索引并合并
        while not result_queue.empty():
            sub_idx = result_queue.get()
            for word, doc_set in sub_idx.items():
                if word not in self.index:
                    self.index[word] = set()
                self.index[word].update(doc_set)

    @staticmethod
    def oneProcessTask(listOfDoc, result_queue):
        tempDict = defaultdict(set)
        for name in listOfDoc:
            with open(name) as f:
                text = f.read()
                words = re.findall(r'\b\w+\b', text.lower())
                for w in words:
                    tempDict[w].add(str(name))
        # 将子字典放入队列
        result_queue.put(dict(tempDict))

    def getListOfDoc(self, keyWord):
        return self.index.get(keyWord.lower(), set())


if __name__ == '__main__':
    ii = InvertedIndex()
    start_time = time.time()
    ii.createIndex()
    print("--- %s seconds ---" % (time.time() - start_time))

关键改动说明： #

1. 创建mp.Queue用于子进程向主进程传递结果
2. 修改oneProcessTask参数，增加队列并在任务完成后放入结果
3. 主进程在所有子进程结束后，从队列取出所有子字典并合并

额外性能优化建议 #

• 批量读取文件：如果文件数量极大，可以考虑使用concurrent.futures配合文件批量读取，减少IO等待
• 词法分析优化：使用更高效的分词库（比如nltk）替代正则表达式，提升单词提取速度
• 内存优化：如果索引过大，可以考虑将部分中间结果写入磁盘，避免内存溢出

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