快捷導(dǎo)航

Python進(jìn)程池Pool應(yīng)用實(shí)例分析

更新時(shí)間：2019年11月27日 09:28:35 作者：xlengji

這篇文章主要介紹了Python進(jìn)程池Pool應(yīng)用,結(jié)合實(shí)例形式分析了Python進(jìn)程池Pool功能、使用方法及相關(guān)操作注意事項(xiàng),需要的朋友可以參考下

本文實(shí)例講述了Python進(jìn)程池Pool應(yīng)用。分享給大家供大家參考，具體如下：

當(dāng)需要?jiǎng)?chuàng)建的子進(jìn)程數(shù)量不多時(shí)，可以直接利用multiprocessing中的Process動(dòng)態(tài)成生多個(gè)進(jìn)程，但如果是上百甚至上千個(gè)目標(biāo)，手動(dòng)的去創(chuàng)建進(jìn)程的工作量巨大，此時(shí)就可以用到multiprocessing模塊提供的Pool方法。

初始化Pool時(shí)，可以指定一個(gè)最大進(jìn)程數(shù)，當(dāng)有新的請(qǐng)求提交到Pool中時(shí)，如果池還沒有滿，那么就會(huì)創(chuàng)建一個(gè)新的進(jìn)程用來執(zhí)行該請(qǐng)求；但如果池中的進(jìn)程數(shù)已經(jīng)達(dá)到指定的最大值，那么該請(qǐng)求就會(huì)等待，直到池中有進(jìn)程結(jié)束，才會(huì)用之前的進(jìn)程來執(zhí)行新的任務(wù)，請(qǐng)看下面的實(shí)例：

# -*- coding:utf-8 -*-
from multiprocessing import Pool
import os, time, random
def worker(msg):
  t_start = time.time()
  print("%s開始執(zhí)行,進(jìn)程號(hào)為%d" % (msg,os.getpid()))
  # random.random()隨機(jī)生成0~1之間的浮點(diǎn)數(shù)
  time.sleep(random.random()*2) 
  t_stop = time.time()
  print(msg,"執(zhí)行完畢，耗時(shí)%0.2f" % (t_stop-t_start))
po = Pool(3) # 定義一個(gè)進(jìn)程池，最大進(jìn)程數(shù)3
for i in range(0,10):
  # Pool().apply_async(要調(diào)用的目標(biāo),(傳遞給目標(biāo)的參數(shù)元祖,))
  # 每次循環(huán)將會(huì)用空閑出來的子進(jìn)程去調(diào)用目標(biāo)
  po.apply_async(worker,(i,))
print("----start----")
po.close() # 關(guān)閉進(jìn)程池，關(guān)閉后po不再接收新的請(qǐng)求
po.join() # 等待po中所有子進(jìn)程執(zhí)行完成，必須放在close語句之后
print("-----end-----")

運(yùn)行結(jié)果:

----start----
0開始執(zhí)行,進(jìn)程號(hào)為21466
1開始執(zhí)行,進(jìn)程號(hào)為21468
2開始執(zhí)行,進(jìn)程號(hào)為21467
0 執(zhí)行完畢，耗時(shí)1.01
3開始執(zhí)行,進(jìn)程號(hào)為21466
2 執(zhí)行完畢，耗時(shí)1.24
4開始執(zhí)行,進(jìn)程號(hào)為21467
3 執(zhí)行完畢，耗時(shí)0.56
5開始執(zhí)行,進(jìn)程號(hào)為21466
1 執(zhí)行完畢，耗時(shí)1.68
6開始執(zhí)行,進(jìn)程號(hào)為21468
4 執(zhí)行完畢，耗時(shí)0.67
7開始執(zhí)行,進(jìn)程號(hào)為21467
5 執(zhí)行完畢，耗時(shí)0.83
8開始執(zhí)行,進(jìn)程號(hào)為21466
6 執(zhí)行完畢，耗時(shí)0.75
9開始執(zhí)行,進(jìn)程號(hào)為21468
7 執(zhí)行完畢，耗時(shí)1.03
8 執(zhí)行完畢，耗時(shí)1.05
9 執(zhí)行完畢，耗時(shí)1.69
-----end-----

multiprocessing.Pool常用函數(shù)解析：

apply_async(func[, args[, kwds]]) ：使用非阻塞方式調(diào)用func（并行執(zhí)行，堵塞方式必須等待上一個(gè)進(jìn)程退出才能執(zhí)行下一個(gè)進(jìn)程），args為傳遞給func的參數(shù)列表，kwds為傳遞給func的關(guān)鍵字參數(shù)列表；
close()：關(guān)閉Pool，使其不再接受新的任務(wù)；
terminate()：不管任務(wù)是否完成，立即終止；
join()：主進(jìn)程阻塞，等待子進(jìn)程的退出，必須在close或terminate之后使用；

進(jìn)程池中的Queue

如果要使用Pool創(chuàng)建進(jìn)程，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()，否則會(huì)得到一條如下的錯(cuò)誤信息：

RuntimeError: Queue objects should only be shared between processes through inheritance.

下面的實(shí)例演示了進(jìn)程池中的進(jìn)程如何通信：

# -*- coding:utf-8 -*-
# 修改import中的Queue為Manager
from multiprocessing import Manager,Pool
import os,time,random
def reader(q):
  print("reader啟動(dòng)(%s),父進(jìn)程為(%s)" % (os.getpid(), os.getppid()))
  for i in range(q.qsize()):
    print("reader從Queue獲取到消息：%s" % q.get(True))
def writer(q):
  print("writer啟動(dòng)(%s),父進(jìn)程為(%s)" % (os.getpid(), os.getppid()))
  for i in "itcast":
    q.put(i)
if __name__=="__main__":
  print("(%s) start" % os.getpid())
  q = Manager().Queue() # 使用Manager中的Queue
  po = Pool()
  po.apply_async(writer, (q,))
  time.sleep(1) # 先讓上面的任務(wù)向Queue存入數(shù)據(jù)，然后再讓下面的任務(wù)開始從中取數(shù)據(jù)
  po.apply_async(reader, (q,))
  po.close()
  po.join()
  print("(%s) End" % os.getpid())

運(yùn)行結(jié)果:

(11095) start
writer啟動(dòng)(11097),父進(jìn)程為(11095)
reader啟動(dòng)(11098),父進(jìn)程為(11095)
reader從Queue獲取到消息：i
reader從Queue獲取到消息：t
reader從Queue獲取到消息：c
reader從Queue獲取到消息：a
reader從Queue獲取到消息：s
reader從Queue獲取到消息：t
(11095) End