當需要建立的子程序數量不多時,可以直接利用multiprocessing中的Process動态成生多個程序,但如果是上百甚至上千個目标,手動的去建立程序的工作量巨大,此時就可以用到multiprocessing子產品提供的Pool方法。
初始化Pool時,可以指定一個最大程序數,當有新的請求送出到Pool中時,如果池還沒有滿,那麼就會建立一個新的程序用來執行該請求;但如果池中的程序數已經達到指定的最大值,那麼該請求就會等待,直到池中有程序結束,才會用之前的程序來執行新的任務,請看下面的執行個體:
from multiprocessing import Pool
import os, time, random
def worker(msg):
t_start = time.time()
print("%s開始執行,程序号為%d" % (msg,os.getpid()))
# random.random()随機生成0~1之間的浮點數
time.sleep(random.random()*2)
t_stop = time.time()
print(msg,"執行完畢,耗時%0.2f" % (t_stop-t_start))
po = Pool(3) # 定義一個程序池,最大程序數3
for i in range(0,10):
# Pool().apply_async(要調用的目标,(傳遞給目标的參數元祖,))
# 每次循環将會用空閑出來的子程序去調用目标
po.apply_async(worker,(i,))
print("----start----")
po.close() # 關閉程序池,關閉後po不再接收新的請求
po.join() # 等待po中所
- multiprocessing.Pool常用函數解析:
- apply_async(func[, args[, kwds]]) :使用非阻塞方式調用func(并行執行,堵塞方式必須等待上一個程序退出才能執行下一個程序),args為傳遞給func的參數清單,kwds為傳遞給func的關鍵字參數清單;
- close():關閉Pool,使其不再接受新的任務;
- join():主程序阻塞,等待子程序的退出, 必須在close或terminate之後使用
# -*- coding:utf-8 -*-
# 修改import中的Queue為Manager
from multiprocessing import Manager,Pool
import os,time,random
def reader(q):
print("reader啟動(%s),父程序為(%s)" % (os.getpid(), os.getppid()))
for i in range(q.qsize()):
print("reader從Queue擷取到消息:%s" % q.get(True))
def writer(q):
print("writer啟動(%s),父程序為(%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) # 先讓上面的任務向Queue存入資料,然後再讓下面的任務開始從中取資料
po.apply_async(reader, (q,))
po.close()
po.join()
print("(%s) End" % os.getpid())