程序池Pool

當需要建立的子程序數量不多時，可以直接利用multiprocessing中的Process動態成生多個程序，但如果是上百甚至上千個目標，手動的去建立程序的工作量巨大，此時就可以用到multiprocessing模組提供的Pool方法。

初始化Pool時，可以指定一個最大程序數，當有新的請求提交到Pool中時，如果池還沒有滿，那麼就會建立一個新的程序用來執行該請求；但如果池中的程序數已經達到指定的最大值，那麼該請求就會等待，直到池中有程序結束，才會用之前的程序來執行新的任務，請看下面的例項：

# -*- coding:utf-8 -*-
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中所有子程序執行完成，必須放在close語句之後
print("-----end-----")

執行結果:

----start----
0開始執行,程序號為21466
1開始執行,程序號為21468
2開始執行,程序號為21467
 

0 執行完畢，耗時1.01
3開始執行,程序號為21466
2 執行完畢，耗時1.24
4開始執行,程序號為21467
3 執行完畢，耗時0.56
5開始執行,程序號為21466
1 執行完畢，耗時1.68
6開始執行,程序號為21468
4 執行完畢，耗時0.67
7開始執行,程序號為21467
5 執行完畢，耗時0.83
8開始執行,程序號為21466
6 執行完畢，耗時0.75
9開始執行,程序號為21468
7 執行完畢，耗時1.03
8 執行完畢，耗時1.05
9 執行完畢，耗時1.69
-----end-----

multiprocessing.Pool常用函式解析：

• apply_async(func[, args[, kwds]]) ：使用非阻塞方式呼叫func（並行執行，堵塞方式必須等待上一個程序退出才能執行下一個程序），args為傳遞給func的引數列表，kwds為傳遞給func的關鍵字引數列表；
• close()：關閉Pool，使其不再接受新的任務；
• terminate()：不管任務是否完成，立即終止；
• join()：主程序阻塞，等待子程序的退出， 必須在close或terminate之後使用；

程序池中的Queue

如果要使用Pool建立程序，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()，否則會得到一條如下的錯誤資訊：

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

下面的例項演示了程序池中的程序如何通訊：

# -*- 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())

執行結果:

(11095) start
writer啟動(11097),父程序為(11095)
reader啟動(11098),父程序為(11095)
reader從Queue獲取到訊息：i
reader從Queue獲取到訊息：t
reader從Queue獲取到訊息：c
reader從Queue獲取到訊息：a
reader從Queue獲取到訊息：s
reader從Queue獲取到訊息：t
(11095) End