我一直在使用一组工作人员来实现 python 的多处理库。我实现了以下代码
import main1
t1 = time.time()
p = Pool(cores)
result = p.map(main1, client_list[client])
if result == []:
return []
p.close()
p.join()
print "Time taken in performing request:: ", time.time()-t1
return shorted(result)
但是,运行该进程一段时间后,我的应用程序有很多正在运行的后台进程。这是为我的应用程序执行 ps aux 后的快照
现在,我在 stackoverflow 上读到了很多类似的问题,比如如何杀死多处理模块创建的僵尸进程?这需要使用我已经实现的 .join() ,并且我学会了如何从这里终止所有这些进程Python 多处理杀死进程。但我想知道我的代码可能会出现什么问题。
我无法共享 main1 函数中的所有代码,但我已将整个代码块放入 try catch 块中,以避免主代码中的错误可能导致僵尸进程。
def main1((param1, param2, param3)):
try:
resout.append(some_data) //resout in case of no error
except:
print traceback.format_exc()
resout = [] //sending empty resout in case of error
return resout
我对并行编程的概念仍然很陌生,并且调试问题变得很棘手。任何帮助将不胜感激。
通常最常见的问题是池已创建但未关闭。
我知道保证池关闭的最好方法是使用 try/finally 子句:
try:
pool = Pool(ncores)
pool.map(yourfunction, arguments)
finally:
pool.close()
pool.join()
如果你不想挣扎multiprocessing
,我写了一个简单的包,名为parmap
它封装了多重处理,让我的生活(也可能是你的)更轻松。
pip install parmap
import parmap
parmap.map(yourfunction, arguments)
从 parmap 使用部分:
-
简单的并行示例:
import parmap
y1 = [myfunction(x, argument1, argument2) for x in mylist]
y2 = parmap.map(myfunction, mylist, argument1, argument2)
y1 == y2
-
迭代元组列表:
# You want to do:
z = [myfunction(x, y, argument1, argument2) for (x,y) in mylist]
z = parmap.starmap(myfunction, mylist, argument1, argument2)
# You want to do:
listx = [1, 2, 3, 4, 5, 6]
listy = [2, 3, 4, 5, 6, 7]
param = 3.14
param2 = 42
listz = []
for (x, y) in zip(listx, listy):
listz.append(myfunction(x, y, param1, param2))
# In parallel:
listz = parmap.starmap(myfunction, zip(listx, listy), param1, param2)
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)