首页 文章

python多处理apply_async只使用一个进程

提问于
浏览 次
18

我有一个脚本，包括从列表中打开文件，然后对该文件中的文本执行某些操作 . 我正在使用python多处理和Pool来尝试并行化这个操作 . 脚本的抽象如下：

import os
from multiprocessing import Pool

results = []
def testFunc(files):
    for file in files:
        print "Working in Process #%d" % (os.getpid())
        #This is just an illustration of some logic. This is not what I'm actually doing.
        for line in file:
            if 'dog' in line:
                results.append(line)

if __name__=="__main__":
    p = Pool(processes=2)
    files = ['/path/to/file1.txt', '/path/to/file2.txt']
    results = p.apply_async(testFunc, args = (files,))
    results2 = results.get()

当我运行它时，每次迭代的进程id的打印输出都是相同的 . 基本上我正在尝试做的是获取输入列表的每个元素并将其分支到一个单独的进程，但似乎一个进程正在完成所有工作 .

python multiprocessing

2 回答

  • 28

    • apply_async 将一项任务提供给池 . 您需要多次调用 apply_async 才能运行更多处理器 .

    • 不允许两个进程尝试写入同一列表 results . 由于池工作者是单独的进程，因此两者不会写入同一列表 . 解决此问题的一种方法是使用输出队列 . 您可以自己设置，或使用 apply_async 的回调为您设置队列 . apply_async 将在函数完成后调用回调 .

    • 您可以使用 map_async 而不是 apply_async ，但是然后您必须展平'd get a list of lists, which you' .

    所以，也许尝试改为：

    import os
import multiprocessing as mp

results = []   

def testFunc(file):
    result = []
    print "Working in Process #%d" % (os.getpid())
    # This is just an illustration of some logic. This is not what I'm
    # actually doing.
    with open(file, 'r') as f:
        for line in f:
            if 'dog' in line:
                result.append(line)
    return result


def collect_results(result):
    results.extend(result)

if __name__ == "__main__":
    p = mp.Pool(processes=2)
    files = ['/path/to/file1.txt', '/path/to/file2.txt']
    for f in files:
        p.apply_async(testFunc, args=(f, ), callback=collect_results)
    p.close()
    p.join()
    print(results)
    回复于
  • 7

    也许在这种情况下你应该使用 map_async ：

    import os
from multiprocessing import Pool

results = []
def testFunc(file):
    message =  ("Working in Process #%d" % (os.getpid()))
    #This is just an illustration of some logic. This is not what I'm actually doing.
    for line in file:
        if 'dog' in line:
            results.append(line)
    return message

if __name__=="__main__":
    print("saddsf")
    p = Pool(processes=2)
    files = ['/path/to/file1.txt', '/path/to/file2.txt']
    results = p.map_async(testFunc, files)
    print(results.get())
    回复于

相关问题