【python】多进程返回值比较 python

every blog every motto: Light tomorrow with today.
0. 前言网上目前关于多进程返回值的文章较为零散，本文主要进行简单的小结。
说明：

其中被测试函数的函数返回值作为函数参数。所以使用多进程运行时间并没有减少，反而更慢，这是需要说明的，关于运行时间，仅作一般结果进行展示，不是本文的重点。
其中关于apply和apply_async两种方法运行时间的比较可参考时间比较

1. 正文测试函数时间用到装饰器，具体可参考装饰器
【【python】多进程返回值比较】导入模块：

import threading from queue import Queue import multiprocessing from multiprocessing import Manager import timeloop_number = 30000 # 循环次数

1.1 无多进程情况

def fun(k): """被测试函数""" print('-----fun函数，参数为{}----'.format(k)) m = k + 10 return m

@count_time def call_fun(): """没有使用多线/进程的情况""" number = 0for i in range(loop_number): print('number:{}'.format(number)) number = fun(number)def main(): """主程序"""# 1. 没有使用多线程 call_fun()if __name__ == '__main__': main()

结果：

文章图片

1.2 多进程返回值 1.2.1 方法一：进程池Pool
有关多种进程池的apply，apply_async的时间比较可参考文献1
1. apply

def fun(k): """被测试函数""" print('-----fun函数，参数为{}----'.format(k)) m = k + 10 return m@count_time def my_process(): """多进程"""# 方法一：apply/apply_async pool = multiprocessing.Pool(4)# 创建4个进程 k = 0 for i in range(loop_number): k = pool.apply(fun, args=(k,) print('返回值为:', k) def main(): """主程序"""# 3. 使用多进程 my_process()if __name__ == '__main__': main()

结果：

文章图片

2. apply_async

def fun(k): """被测试函数""" print('-----fun函数，参数为{}----'.format(k)) m = k + 10 return m@count_time def my_process(): """多进程"""# 方法一：apply/apply_async pool = multiprocessing.Pool(4)# 创建4个进程 k = 0 for i in range(loop_number): k = pool.apply_async(fun, args=(k,)) k = k.get() print('返回值为:', k) def main(): """主程序"""# 3. 使用多进程 my_process()if __name__ == '__main__': main()

结果：

文章图片

说明：
上述运行时间，仅作一般参考，运行时间非本文重点，具体参考文献1
小结：

因为测试函数的特殊性（函数返回值作为参数传递进函数），两种方法都相当于串行，所以时间方面并没差距
apply_async方法返回结果为对象，所以需要用get方法获取结果

apply_async关于返回结果位置小讨论： 1. 返回结果中间输出：

@count_time def my_process(): """多进程"""# 方法一：apply/apply_async pool = multiprocessing.Pool(4)# 创建4个进程 k = 0 for i in range(loop_number): t = pool.apply_async(fun, args=(k,))print('返回值为:', t.get())

结果：

文章图片

2. 返回结果最后输出：

@count_time def my_process(): """多进程"""# 方法一：apply/apply_async pool = multiprocessing.Pool(4)# 创建4个进程 k = 0 li = [] # 空列表 for i in range(loop_number): t = pool.apply_async(fun, args=(k,)) li.append(t) for i in li: print('返回值为:', i.get())

结果：

文章图片

时间不同的原因：

在进行得到各个结果的时候，是将结果追加到列表中，否则，在得到结果get的时候会阻塞进程，从而将多进程变成了单进程，当使用列表来存放时，在进行get数据的时候，可以设置超时时间，如，get(timeout=5)。

小结：

对于apply_async返回结果最后输出时间更短
对于apply二者没有明显区别，读者可自行验证

1.2.2 方法二：Manger

from multiprocessing import Manager, Processdef fun(k,result_dict): """被测试函数""" print('-----fun函数内部，参数为{}----'.format(k)) m = k + 10result_dict[k] = m # 方法二：manger@count_time def my_process(): """多进程"""# 方法二：Manger manger = Manager() result_dict = manger.dict()# 使用字典 jobs = []for i in range(10): p = Process(target=fun, args=(i, result_dict)) jobs.append(p) p.start()for pr in jobs: pr.join() var = result_dict print('返回结果', var.values())def main(): """主程序"""# 3. 使用多进程 my_process()if __name__ == '__main__': main()

结果：

文章图片

说明：

上面用的是字典，也可以用列表
返回的结果只有等所有子进程运行完以后才能获取，从这一点上来说，不满足我们的要求，由于作为获取多进程的一种方法，附上。

1.2.3 方法三：Pipe
定义类

class MyProcess(multiprocessing.Process): def __init__(self, name, func, args): super(MyProcess, self).__init__() self.name = name self.func = func self.args = args self.res = ''def run(self): self.res = self.func(*self.args)

def fun(k,p): """被测试函数""" print('-----fun函数内部，参数为{}----'.format(k)) m = k + 10p.send(m)@count_time def my_process(): """多进程"""# 方法三： process_li = [] parent_con, child_con = multiprocessing.Pipe() for i in range(30): p = MyProcess('proce', fun, (1, child_con)) process_li.append(p) p.start() for i in process_li: i.join() for i in process_li: print(parent_con.recv())

结果：

文章图片

说明：

当参数loop_number=30000时，内存溢出，电脑被搞黑屏了，所以仅循环30次，
函数返回的结果只能最后一起输出，不符合要求。

1.2.3 方法四：Quene
定义类

class MyProcess(multiprocessing.Process): def __init__(self, name, func, args): super(MyProcess, self).__init__() self.name = name self.func = func self.args = args self.res = ''def run(self): self.res = self.func(*self.args)

def fun(k, p): """被测试函数""" print('-----fun函数内部，参数为{}----'.format(k)) m = k + 10p.put(m)@count_time def my_process(): """多进程"""# 方法四：Queue process_li = [] q = multiprocessing.Queue() k = 1 for i in range(30): p = MyProcess('proce', fun, (k, q)) p.start() process_li.append(p) for i in process_li: i.join()while q.qsize() > 0: print(q.get())

结果：

文章图片

说明：

和上面Pipe一样，内存溢出，其结果只能放在队列当中，只能最后取得，不满足需求

总结：

建议使用进程池apply / apply_async
参照这篇文章apply/apply_async时间比较分析，推荐使用ThreadPool下的apply_async

参考文献 [1] https://blog.csdn.net/weixin_39190382/article/details/107865552
[2] https://zhidao.baidu.com/question/246068963774450124.html
[3] https://blog.csdn.net/sunt2018/article/details/85336408
[4] https://blog.csdn.net/ztf312/article/details/80337255
[5] https://www.jb51.net/article/86412.htm
[6] https://blog.csdn.net/littlehaes/article/details/102626610