python的multiprocess的使用

主要参考链接&＃xff1a;

1&＃xff09;http://www.jb51.net/article/67116.htm

2&＃xff09;http://www.cnblogs.com/kaituorensheng/p/4445418.html

3&＃xff09;https://jingsam.github.io/2015/12/31/multiprocessing.html      写的很好

4)http://www.cnblogs.com/vamei/archive/2012/10/12/2721484.html

一。http://www.jb51.net/article/67116.htm的摘要

由于Python设计的限制(我说的是咱们常用的CPython)。最多只能用满1个CPU核心。这就是导致在做特征工程的时候&＃xff0c;运行效率慢的主要原因&＃xff0c;拿一个cpu单线程跑特征工程肯定慢。
Python提供了非常好用的多进程包multiprocessing&＃xff0c;你只需要定义一个函数&＃xff0c;Python会替你完成其他所有事情。借助这个包&＃xff0c;可以轻松完成从单进程到并发执行的转换。

1、新建单一进程

如果我们新建少量进程&＃xff0c;可以如下&＃xff1a;

2、使用进程池

是的&＃xff0c;你没有看错&＃xff0c;不是线程池。它可以让你跑满多核CPU&＃xff0c;而且使用方法非常简单。

注意要用apply_async&＃xff0c;如果落下async&＃xff0c;就变成阻塞版本了&＃xff0c;apply&＃xff08;&＃xff09;这个表示上一个线程结束之后才能进行下一个任务&＃xff0c;也就是前面所说的阻塞。如果是apply_async()&＃xff0c;那么就可以异步执行。

processes&＃61;4是最多并发进程数量。

3、使用Pool&＃xff0c;并需要关注结果

更多的时候&＃xff0c;我们不仅需要多进程执行&＃xff0c;还需要关注每个进程的执行结果&＃xff0c;如下&＃xff1a;

python中的多线程其实并不是真正的多线程&＃xff0c;如果想要充分地使用多核CPU的资源&＃xff0c;在python中大部分情况需要使用多进程。Python提供了非常好用的多进程包multiprocessing&＃xff0c;只需要定义一个函数&＃xff0c;Python会完成其他所有事情。借助这个包&＃xff0c;可以轻松完成从单进程到并发执行的转换。multiprocessing支持子进程、通信和共享数据、执行不同形式的同步&＃xff0c;提供了Process、Queue、Pipe、Lock等组件。

对于pool的讲解&＃xff1a;

Pool

在利用Python进行系统管理的时候&＃xff0c;特别是同时操作多个文件目录&＃xff0c;或者远程控制多台主机&＃xff0c;并行操作可以节约大量的时间。当被操作对象数目不大时&＃xff0c;可以直接利用multiprocessing中的Process动态成生多个进程&＃xff0c;十几个还好&＃xff0c;但如果是上百个&＃xff0c;上千个目标&＃xff0c;手动的去限制进程数量却又太过繁琐&＃xff0c;此时可以发挥进程池的功效。
Pool可以提供指定数量的进程&＃xff0c;供用户调用&＃xff0c;当有新的请求提交到pool中时&＃xff0c;如果池还没有满&＃xff0c;那么就会创建一个新的进程用来执行该请求&＃xff1b;但如果池中的进程数已经达到规定最大值&＃xff0c;那么该请求就会等待&＃xff0c;直到池中有进程结束&＃xff0c;才会创建新的进程来它。

例7.1&＃xff1a;使用进程池&＃xff08;非阻塞&＃xff09;

#coding: utf-8
import multiprocessing
import timedef func(msg):print "msg:", msgtime.sleep(3)print "end"if __name__ &＃61;&＃61; "__main__":pool &＃61; multiprocessing.Pool(processes &＃61; 3)for i in xrange(4):msg &＃61; "hello %d" %(i)pool.apply_async(func, (msg, )) #维持执行的进程总数为processes&＃xff0c;当一个进程执行完毕后会添加新的进程进去print "Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~"pool.close()pool.join() #调用join之前&＃xff0c;先调用close函数&＃xff0c;否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束print "Sub-process(es) done."

一次执行结果

函数解释&＃xff1a;

• apply_async(func[, args[, kwds[, callback]]]) 它是非阻塞&＃xff0c;apply(func[, args[, kwds]])是阻塞的&＃xff08;理解区别&＃xff0c;看例1例2结果区别&＃xff09;
• close()    关闭pool&＃xff0c;使其不在接受新的任务。
• terminate()    结束工作进程&＃xff0c;不在处理未完成的任务。
• join()    主进程阻塞&＃xff0c;等待子进程的退出&＃xff0c; join方法要在close或terminate之后使用。

执行说明&＃xff1a;创建一个进程池pool&＃xff0c;并设定进程的数量为3&＃xff0c;xrange(4)会相继产生四个对象[0, 1, 2, 4]&＃xff0c;四个对象被提交到pool中&＃xff0c;因pool指定进程数为3&＃xff0c;所以0、1、2会直接送到进程中执行&＃xff0c;当其中一个执行完事后才空出一个进程处理对象3&＃xff0c;所以会出现输出“msg: hello 3”出现在"end"后。因为为非阻塞&＃xff0c;主函数会自己执行自个的&＃xff0c;不搭理进程的执行&＃xff0c;所以运行完for循环后直接输出“mMsg: hark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~”&＃xff0c;主程序在pool.join&＃xff08;&＃xff09;处等待各个进程的结束。

例7.2&＃xff1a;使用进程池&＃xff08;阻塞&＃xff09;

#coding: utf-8
import multiprocessing
import timedef func(msg):print "msg:", msgtime.sleep(3)print "end"if __name__ &＃61;&＃61; "__main__":pool &＃61; multiprocessing.Pool(processes &＃61; 3)for i in xrange(4):msg &＃61; "hello %d" %(i)pool.apply(func, (msg, )) #维持执行的进程总数为processes&＃xff0c;当一个进程执行完毕后会添加新的进程进去print "Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~"pool.close()pool.join() #调用join之前&＃xff0c;先调用close函数&＃xff0c;否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束print "Sub-process(es) done."

一次执行的结果

例7.3&＃xff1a;使用进程池&＃xff0c;并关注结果

import multiprocessing
import timedef func(msg):print "msg:", msgtime.sleep(3)print "end"return "done" &＃43; msgif __name__ &＃61;&＃61; "__main__":pool &＃61; multiprocessing.Pool(processes&＃61;4)result &＃61; []for i in xrange(3):msg &＃61; "hello %d" %(i)result.append(pool.apply_async(func, (msg, )))pool.close()pool.join()for res in result:print ":::", res.get()print "Sub-process(es) done."

一次执行结果

例7.4&＃xff1a;使用多个进程池

#coding: utf-8
import multiprocessing
import os, time, randomdef Lee():print "\nRun task Lee-%s" %(os.getpid()) #os.getpid()获取当前的进程的IDstart &＃61; time.time()time.sleep(random.random() * 10) #random.random()随机生成0-1之间的小数end &＃61; time.time()print &＃39;Task Lee, runs %0.2f seconds.&＃39; %(end - start)def Marlon():print "\nRun task Marlon-%s" %(os.getpid())start &＃61; time.time()time.sleep(random.random() * 40)end&＃61;time.time()print &＃39;Task Marlon runs %0.2f seconds.&＃39; %(end - start)def Allen():print "\nRun task Allen-%s" %(os.getpid())start &＃61; time.time()time.sleep(random.random() * 30)end &＃61; time.time()print &＃39;Task Allen runs %0.2f seconds.&＃39; %(end - start)def Frank():print "\nRun task Frank-%s" %(os.getpid())start &＃61; time.time()time.sleep(random.random() * 20)end &＃61; time.time()print &＃39;Task Frank runs %0.2f seconds.&＃39; %(end - start)if __name__&＃61;&＃61;&＃39;__main__&＃39;:function_list&＃61; [Lee, Marlon, Allen, Frank] print "parent process %s" %(os.getpid())pool&＃61;multiprocessing.Pool(4)for func in function_list:pool.apply_async(func) #Pool执行函数&＃xff0c;apply执行函数,当有一个进程执行完毕后&＃xff0c;会添加一个新的进程到pool中print &＃39;Waiting for all subprocesses done...&＃39;pool.close()pool.join() #调用join之前&＃xff0c;一定要先调用close() 函数&＃xff0c;否则会出错, close()执行后不会有新的进程加入到pool,join函数等待素有子进程结束print &＃39;All subprocesses done.&＃39;

一次执行结果

三。https://jingsam.github.io/2015/12/31/multiprocessing.html

pool.apply_async 采用异步方式调用 task&＃xff0c;pool.apply 则是同步方式调用。同步方式意味着下一个 task 需要等待上一个 task 完成后才能开始运行&＃xff0c;这显然不是我们想要的功能&＃xff0c;所以采用异步方式连续地提交任务。在上面的语句中&＃xff0c;我们提交了 4 个任务&＃xff0c;假设我的 CPU 是 4 核&＃xff0c;那么我的每个核运行一个任务。如果我提交多于 4 个任务&＃xff0c;那么每个核就需要同时运行 2 个以上的任务&＃xff0c;这回带来任务切换成本&＃xff0c;降低了效率。所以我们设置的并行任务数最好等于 CPU 核心数&＃xff0c; CPU 核可以通过下面语句得到&＃xff1a;

接下来我们使用 result.get() 来获取 task 的返回值&＃xff1a;

在这里不免有人要疑问&＃xff0c;为什么不直接在 for 循环中直接 result.get()呢&＃xff1f;这是因为pool.apply_async之后的语句都是阻塞执行的&＃xff0c;调用 result.get() 会等待上一个任务执行完之后才会分配下一个任务。事实上&＃xff0c;获取返回值的过程最好放在进程池回收之后进行&＃xff0c;避免阻塞后面的语句。

最后我们使用一下语句回收进程池&＃xff1a;

最后附上完整的代码如下&＃xff1a;