在下面的示例代码中,我想获取函数worker的返回值。我该怎么做呢?这个值存储在哪里?

示例代码:

import multiprocessing

def worker(procnum):
    '''worker function'''
    print str(procnum) + ' represent!'
    return procnum


if __name__ == '__main__':
    jobs = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i,))
        jobs.append(p)
        p.start()

    for proc in jobs:
        proc.join()
    print jobs

输出:

0 represent!
1 represent!
2 represent!
3 represent!
4 represent!
[<Process(Process-1, stopped)>, <Process(Process-2, stopped)>, <Process(Process-3, stopped)>, <Process(Process-4, stopped)>, <Process(Process-5, stopped)>]

我似乎无法在存储在作业中的对象中找到相关属性。


当前回答

pebble包有一个很好的利用多处理的抽象。管道,这使得这个非常简单:

from pebble import concurrent

@concurrent.process
def function(arg, kwarg=0):
    return arg + kwarg

future = function(1, kwarg=1)

print(future.result())

示例来自:https://pythonhosted.org/Pebble/#concurrent-decorators

其他回答

使用共享变量进行通信。比如这样:

import multiprocessing


def worker(procnum, return_dict):
    """worker function"""
    print(str(procnum) + " represent!")
    return_dict[procnum] = procnum


if __name__ == "__main__":
    manager = multiprocessing.Manager()
    return_dict = manager.dict()
    jobs = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i, return_dict))
        jobs.append(p)
        p.start()

    for proc in jobs:
        proc.join()
    print(return_dict.values())

我想我应该简化上面复制的最简单的例子,在Py3.6上为我工作。最简单的是多处理。池:

import multiprocessing
import time

def worker(x):
    time.sleep(1)
    return x

pool = multiprocessing.Pool()
print(pool.map(worker, range(10)))

您可以设置池中的进程数,例如,pool (processes=5)。但是,它默认为CPU计数,因此对于CPU受限的任务,请将其保留为空。(I/ o绑定的任务通常适合线程,因为线程大部分都在等待,所以可以共享一个CPU内核。)Pool还应用了分块优化。

(注意,工作方法不能嵌套在方法中。我最初在调用池的方法中定义了工作方法。map,以保持它完全自包含,但随后进程无法导入它,并抛出“AttributeError: Can't pickle local object outer_method..inner_method”。更多的在这里。它可以在类内部。)

(欣赏原文问题指定打印’代表!'而不是time.sleep(),但没有它,我认为一些代码是并发运行时,它不是。)


Py3的ProcessPoolExecutor也是两行(。Map返回一个生成器,所以你需要list()):

from concurrent.futures import ProcessPoolExecutor
with ProcessPoolExecutor() as executor:
    print(list(executor.map(worker, range(10))))

使用普通流程:

import multiprocessing
import time

def worker(x, queue):
    time.sleep(1)
    queue.put(x)

queue = multiprocessing.SimpleQueue()
tasks = range(10)

for task in tasks:
    multiprocessing.Process(target=worker, args=(task, queue,)).start()

for _ in tasks:
    print(queue.get())

如果您所需要的只是放置和获取,请使用SimpleQueue。在第二个循环进入阻塞队列之前,第一个循环启动所有进程。得到调用。我认为没有任何理由也调用p.join()。

似乎应该使用多处理。使用.apply() .apply_async(), map()方法

http://docs.python.org/library/multiprocessing.html?highlight=pool#multiprocessing.pool.AsyncResult

你可以使用ProcessPoolExecutor从函数中获取一个返回值,如下所示:

from concurrent.futures import ProcessPoolExecutor

def test(num1, num2):
    return num1 + num2

with ProcessPoolExecutor() as executor:
    feature = executor.submit(test, 2, 3)
    print(feature.result()) # 5

pebble包有一个很好的利用多处理的抽象。管道,这使得这个非常简单:

from pebble import concurrent

@concurrent.process
def function(arg, kwarg=0):
    return arg + kwarg

future = function(1, kwarg=1)

print(future.result())

示例来自:https://pythonhosted.org/Pebble/#concurrent-decorators