在下面的示例代码中,我想获取函数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)>]
我似乎无法在存储在作业中的对象中找到相关属性。
对于正在寻找如何使用Queue从进程中获取值的任何人:
import multiprocessing
ret = {'foo': False}
def worker(queue):
ret = queue.get()
ret['foo'] = True
queue.put(ret)
if __name__ == '__main__':
queue = multiprocessing.Queue()
queue.put(ret)
p = multiprocessing.Process(target=worker, args=(queue,))
p.start()
p.join()
print(queue.get()) # Prints {"foo": True}
注意,在Windows或Jupyter Notebook中,使用多线程,您必须将其保存为文件并执行该文件。如果你在命令提示符中这样做,你会看到这样的错误:
AttributeError: Can't get attribute 'worker' on <module '__main__' (built-in)>
这个例子展示了如何使用multiprocessing的列表。管道实例从任意数量的进程中返回字符串:
import multiprocessing
def worker(procnum, send_end):
'''worker function'''
result = str(procnum) + ' represent!'
print result
send_end.send(result)
def main():
jobs = []
pipe_list = []
for i in range(5):
recv_end, send_end = multiprocessing.Pipe(False)
p = multiprocessing.Process(target=worker, args=(i, send_end))
jobs.append(p)
pipe_list.append(recv_end)
p.start()
for proc in jobs:
proc.join()
result_list = [x.recv() for x in pipe_list]
print result_list
if __name__ == '__main__':
main()
输出:
0 represent!
1 represent!
2 represent!
3 represent!
4 represent!
['0 represent!', '1 represent!', '2 represent!', '3 represent!', '4 represent!']
这种解决方案使用的资源比多进程少。使用
一个管道
至少一个锁
一个缓冲区
一个线程
或者多处理。SimpleQueue使用
一个管道
至少一个锁
查看这些类型的源代码是非常有指导意义的。
如果你正在使用Python 3,你可以使用concurrent.futures.ProcessPoolExecutor作为一个方便的抽象:
from concurrent.futures import ProcessPoolExecutor
def worker(procnum):
'''worker function'''
print(str(procnum) + ' represent!')
return procnum
if __name__ == '__main__':
with ProcessPoolExecutor() as executor:
print(list(executor.map(worker, range(5))))
输出:
0 represent!
1 represent!
2 represent!
3 represent!
4 represent!
[0, 1, 2, 3, 4]
我想我应该简化上面复制的最简单的例子,在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()。
我修改了vartec的答案,因为我需要从函数中获得错误代码。(由于vertec ! !这是一个很棒的技巧)
这也可以通过经理来实现。列表,但我认为最好是在字典中,并在其中存储一个列表。这样,我们就可以保留函数和结果,因为我们不能确定列表将被填充的顺序。
from multiprocessing import Process
import time
import datetime
import multiprocessing
def func1(fn, m_list):
print 'func1: starting'
time.sleep(1)
m_list[fn] = "this is the first function"
print 'func1: finishing'
# return "func1" # no need for return since Multiprocess doesnt return it =(
def func2(fn, m_list):
print 'func2: starting'
time.sleep(3)
m_list[fn] = "this is function 2"
print 'func2: finishing'
# return "func2"
def func3(fn, m_list):
print 'func3: starting'
time.sleep(9)
# if fail wont join the rest because it never populate the dict
# or do a try/except to get something in return.
raise ValueError("failed here")
# if we want to get the error in the manager dict we can catch the error
try:
raise ValueError("failed here")
m_list[fn] = "this is third"
except:
m_list[fn] = "this is third and it fail horrible"
# print 'func3: finishing'
# return "func3"
def runInParallel(*fns): # * is to accept any input in list
start_time = datetime.datetime.now()
proc = []
manager = multiprocessing.Manager()
m_list = manager.dict()
for fn in fns:
# print fn
# print dir(fn)
p = Process(target=fn, name=fn.func_name, args=(fn, m_list))
p.start()
proc.append(p)
for p in proc:
p.join() # 5 is the time out
print datetime.datetime.now() - start_time
return m_list, proc
if __name__ == '__main__':
manager, proc = runInParallel(func1, func2, func3)
# print dir(proc[0])
# print proc[0]._name
# print proc[0].name
# print proc[0].exitcode
# here you can check what did fail
for i in proc:
print i.name, i.exitcode # name was set up in the Process line 53
# here will only show the function that worked and where able to populate the
# manager dict
for i, j in manager.items():
print dir(i) # things you can do to the function
print i, j