python 实现多进程

---

 参考链接： 

python中实现多进程的模块：multiprocessing

注意：在windows系统下，要想启动一个子进程，必须把进程相关的内容写在”if __name__ == “__main__” ”，这句话下面。

具体实现模块

1、Process模块

• 实现功能：

创建子进程

• 构造方法：

Process([group [, target [, name [, args [, kwargs]]]]])

group: 线程组，目前还没有实现，库引用中提示必须是None；

target: 要执行的方法；

name: 进程名；

args/kwargs: 要传入方法的参数。

• 实例方法：

is_alive()：返回进程是否在运行。

join([timeout])：阻塞当前上下文环境的进程程，直到调用此方法的进程终止或到达指定的timeout（可选参数）。

start()：进程准备就绪，等待CPU调度。

run()：strat()调用run方法，如果实例进程时未制定传入target，这star执行t默认run()方法。

terminate()：不管任务是否完成，立即停止工作进程。

• 属性：

authkey

daemon：和线程的setDeamon功能一样（将父进程设置为守护进程，当父进程结束时，子进程也结束）。

exitcode(进程在运行时为None、如果为–N，表示被信号N结束）。

name：进程名字。

pid：进程号。

• 例子：

1 import multiprocessing2 3 def job(a,d):4 　　print('aaaaa')5 6 if __name__ == “__main__”:7 　　p1 = multiprocessing.Process(target=job,args=(1,2))8 　　p1.start()9 　　p1.join()

 

2、Pool模块

• 实现功能：

创建管理进程池。提供指定数量的进程供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，才会创建新的进程来执行它。在共享资源时，只能使用Multiprocessing.Manager类，而不能使用Queue或者Array。

• 构造方法：

Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])

processes ：使用的工作进程的数量，如果processes是None那么使用 os.cpu_count()返回的数量。（Pool默认大小是CPU的核数，我们也可以通过在Pool中传入processes参数即可自定义需要的核数量）

initializer： 如果initializer是None，那么每一个工作进程在开始的时候会调用initializer(*initargs)。

maxtasksperchild：工作进程退出之前可以完成的任务数，完成后用一个新的工作进程来替代原进程，来让闲置的资源被释放。maxtasksperchild默认是None，意味着只要Pool存在工作进程就会一直存活。

context: 用在制定工作进程启动时的上下文，一般使用 multiprocessing.Pool() 或者一个context对象的Pool()方法来创建一个池，两种方法都适当的设置了context。

• 实例方法：

apply_async(func[, args[, kwds[, callback]]]) 它是非阻塞。

apply(func[, args[, kwds]])是阻塞的。

close() 关闭pool，使其不在接受新的任务。

terminate() 关闭pool，结束工作进程，不在处理未完成的任务。

join() 主进程阻塞，等待子进程的退出， join方法要在close或terminate之后使用。

• Pool使用方法：

1、Pool+map函数

• 用map()获取结果，在map()中需要放入函数和需要迭代运算的值，然后它会自动分配给CPU核，返回结果。
• 说明：此写法缺点在于只能通过map向函数传递一个参数。

1 from multiprocessing import Pool 2  3 def test(i): 4     print i 5  6 if __name__=="__main__": 7     lists=[1,2,3] 8     pool=Pool(processes=2)   #定义最大的进程数 9     pool.map(test,lists)   #lists必须是一个可迭代变量。10     pool.close()11     pool.join()

2、异步进程池（非阻塞）

• apply_async()只能放入一组参数，并返回一个结果，如果想得到map()的效果需要通过迭代。

1 from multiprocessing import Pool 2  3 def test(i): 4     print i 5  6 if __name__=="__main__": 7     pool = Pool(processes=10) 8     for i in xrange(500): 9         '''10          For循环中执行步骤：11         （1）循环遍历，将500个子进程添加到进程池（相对父进程会阻塞）12         （2）每次执行10个子进程，等一个子进程执行完后，立马启动新的子进程。（相对父进程不阻塞）13         apply_async为异步进程池写法。14         异步指的是启动子进程的过程，与父进程本身的执行（print）是异步的，而For循环中往进程池添加子进程的过程，与父进程本身的执行却是同步的。15         '''16         pool.apply_async(test, args=(i,)) #维持执行的进程总数为10，当一个进程执行完后启动一个新进程. 17     print“test”18     pool.close()19     pool.join()

代码说明：

执行顺序：For循环内执行了2个步骤，第一步：将500个对象放入进程池（阻塞）。第二步：同时执行10个子进程（非阻塞），有结束的就立即添加，维持10个子进程运行。（apply_async方法的会在执行完for循环的添加步骤后，直接执行后面的print语句，而apply方法会等所有进程池中的子进程运行完以后再执行后面的print语句）

注意：调用join之前，先调用close或者terminate方法，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束。

3、同步进程池（阻塞）

1 from multiprocessing import Pool 2  3 def test(p): 4     print p 5     time.sleep(3) 6  7 if __name__=="__main__": 8     pool = Pool(processes=10) 9     for i in xrange(500):10         '''11         实际测试发现，for循环内部执行步骤：12         （1）遍历500个可迭代对象，往进程池放一个子进程13         （2）执行这个子进程，等子进程执行完毕，再往进程池放一个子进程，再执行。（同时只执行一个子进程）14         for循环执行完毕，再执行print函数。15         '''16         pool.apply(test, args=(i,)) #维持执行的进程总数为10，当一个进程执行完后启动一个新进程.17     print“test”18     pool.close()19     pool.join()

代码说明：

for循环内执行的步骤顺序，往进程池中添加一个子进程，执行子进程，等待执行完毕再添加一个子进程…..等500个子进程都执行完了，再执行print “test”。（从结果来看，并没有多进程并发

 

3、Queue模块

• 实现功能：

将每个核或线程的运算结果放在队列中，等到每个线程或核运行完毕后再从队列中取出结果， 继续加载运算。原因很简单, 多线程调用的函数不能有返回值, 所以使用Queue存储多个线程运算的结果。

• 例子：

1 import multiprocessing as mp 2  3 def job(q): 4     res=0 5     for i in range(1000): 6         res+=i+i**2+i**3 7     q.put(res) #queue 8  9 if __name__=='__main__':10     q = mp.Queue()11     p1 = mp.Process(target=job,args=(q,))12     p2 = mp.Process(target=job,args=(q,))13     p1.start()14     p2.start()15     p1.join()16     p2.join()17     res1 = q.get()18     res2 = q.get()19     print(res1+res2)

 

4、Pipe模块

• 实现功能：

用来管道操作。

5、Manager模块

• 实现功能：

Manager模块常与Pool模块一起使用，作用是共享资源。

 

6、Lock模块（进程锁）

• 实现功能：

当多个进程需要访问共享资源的时候，Lock可以用来避免访问的冲突。

• 实现步骤：

1）首先需要定义一个进程锁

l = multiprocessing.Lock()    # 定义一个进程锁

2）然后将进程锁的信息传入各个进程中

p1 = multiprocessing.Process(target=job, args=(v,1,l))    # 需要将Lock传入

p2 = multiprocessing.Process(target=job, args=(v,3,l))

3）在job()中设置进程锁的使用，保证运行时一个进程的对锁内内容的独占

• 例子：

1 import multiprocessing as mp 2  3 def job(v, num, l): 4     l.acquire() # 锁住 5     for _ in range(5): 6         time.sleep(0.1)  7         v.value += num # 获取共享内存 8         print(v.value) 9     l.release() # 释放10 11 def multicore():12     l = mp.Lock() # 定义一个进程锁13     v = mp.Value('i', 0) # 定义共享内存14     p1 = mp.Process(target=job, args=(v,1,l)) # 需要将lock传入15     p2 = mp.Process(target=job, args=(v,3,l)) 16     p1.start()17     p2.start()18     p1.join()19     p2.join()20 21 if __name__ == '__main__':22     multicore()

 

共享内存

Multiprocessing类中共享资源可以使用3种方式，分别是Queue，Array，Manager。

1、Queue类

使用Multiprocessing.Queue类，共享资源（share memory）（只适用Process类，不能再Pool进程池中使用）

1 from multiprocessing import Process, Queue 2  3 def test(queue): 4     queue.put("Hello World") 5  6 if __name__ == '__main__': 7     q = Queue() 8     p = Process(target=test, args=(q,)) #需要将q对象传递给子进程 9     p.start()10     Print q.get()

2、Array、Value类

使用Multiprocessing.Array类，共享资源（share memory）（只适用于Process类，无法与Pool一起使用）

1 from multiprocessing import Process, Array 2  3 def test(a): 4     for i in range(len(a)): 5         a[i] = -a[i] 6  7 if__name__ == '__main__': 8     arr = Array('i', range(10)) 9     p = Process(target=test, args=(arr)) #需要将arr对象传递给子进程10     p.start()11     p.join()12     print arr[:]

• 单值：Value

　　我们可以通过使用Value数据存储在一个共享的内存表中。

1 import multiprocessing as mp　2 value1 = mp.Value('i', 0) 3 value2 = mp.Value('d', 3.14)4 　　5 # 其中d和i参数用来设置数据类型的，d表示一个双精浮点类型，i表示一个带符号的整型。

• 列表：Array

　　在Python的mutiprocessing中，有还有一个Array类，可以和共享内存交互，来实现在进程之间共享数据。

1 array = mp.Array('i', [1, 2, 3, 4])2 　　3 #这里的Array和numpy中的不同，它只能是一维的，不能是多维的。同样和Value 一样，需要定义数据形式，否则会报错。

3、Manager类

使用Multiprocessing.Manager类，共享资源。（可以适用Pool类）

实例目的：父进程在执行子进程的过程中，同步判断一个公共资源值，如果满足条件则结束所有进程。

1 from multiprocessing import Manager 2  3 def test(i,lists): 4     print i 5     lists.append(i) 6  7 if __name__=="__main__": 8     pool=Pool() 9     lists=Manager().list() #Manager类实例化代码只能写在main()函数里面10     for i in xrange(10000000):11         if len(lists)<=0:12             '''13             在创建子进程时，需要将lists对象传入，不然无法共享。14             '''15             pool.apply_async(test,args=(i,lists))##需要将lists对象传递给子进程，这里比较耗资源，原因可能是因为Manager类是基于通信的。16         else:17             break

　　

• 父进程中的全局变量能被子进程共享吗？

解答：不行，因为每个进程享有独立的内存数据，如果想要共享资源，可以使用Manage类，或者Queue等模块。