使用 Multiprocessing 处理长列表

Question

output = mp.Queue()

def endScoreList(all_docs, query, pc, output):
    score_list = []
    for doc in all_docs:
        print "In process", pc
        score_list.append(some_score(doc, query))
        print "size of score_list is", len(score_list)
    output.put((doc, score_list))

if __name__ == '__main__':
    mp.freeze_support()
    num_of_workers = mp.cpu_count()
    doc_list = getDocuments(query)
    ## query is a list of strings.
    ## doc_list is a list of document names
    processes = [mp.Process(target = endScoreList, args = (doc_list, x, query, output)) for x in range(num_of_workers)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    results = [output.get() for p in processes]
    print results

我有一个文档名称列表all_docs，我必须将其数据与输入query 进行比较。这是使用我从some_score(doc, query) 获得的分数来完成的。文档列表约为 100k。我必须得到所有文件的分数。如何制作程序以并行生成分数。分数是相互独立的，所以最后我只需要合并所有返回的(doc, score) 列表。我试着做一个程序，但我不认为它是并行运行的。

请帮帮我。

我使用的是 Windows 64 位/i7。

Answer 1

很难指出您当前的代码出了什么问题，因为您展示的示例存在许多问题（例如，您使用 // 引入注释，创建调用一个finalScore 函数并将doc_list 作为参数传递，两者均未定义）。

与其试图弄清楚你的代码发生了什么，我想提出一个可能更简单的替代解决方案。如果您使用multiprocessing.Pool 的map 方法，无论池中有多少进程，您的工作都会被分配。

import multiprocessing as mp

def worker(doc):
    return doc, some_score(doc, "query")

if __name__ == "__main__":
    mp.freeze_support()
    p = mp.Pool() # default is a number of processes equal to the number of CPU cores
    scores = p.map(worker, all_docs)
    p.close()
    p.join()

这个简单的版本假设查询字符串是一个常量。如果不是这种情况，您可以将其作为参数传递给map 调用（或考虑改用starmap）。