python進程池

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

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

import multiprocessing
import time
import os
import random


def test1(msg):
        t_start = time.time()   
        print("%s開始執行，進程號為%d" % (msg, os.getpid()))
        time.sleep(random.random() * 2)  
        t_stop = time.time()
        print("%s執行完成，耗時%.2f" % (msg, t_stop - t_start)) 


if __name__ == "__main__":
    
        po = multiprocessing.Pool(3)
        for i in range(0, 10):
                # Pool().apply_async(要調用的目標,(傳遞給目標的參數元祖,))
                # 每次循環將會用空閑出來的子進程去調用目標
                po.apply_async(test1, (i,))
    
        print("-----start-----")
    
        po.close() # 關閉進程池，關閉后po不再接收新的請求
        po.join() # 等待po中所有子進程執行完成，必須放在close語句之后

        print("-----end-----")

運行結果：

-----start-----
0開始執行，進程號為40291
1開始執行，進程號為40292
2開始執行，進程號為40293
2執行完成，耗時0.59
3開始執行，進程號為40293
0執行完成，耗時1.21
4開始執行，進程號為40291
1執行完成，耗時1.56
5開始執行，進程號為40292
3執行完成，耗時1.58
6開始執行，進程號為40293
5執行完成，耗時1.36
7開始執行，進程號為40292
4執行完成，耗時1.73
8開始執行，進程號為40291
6執行完成，耗時1.34
9開始執行，進程號為40293
8執行完成，耗時0.71
9執行完成，耗時0.36
7執行完成，耗時1.21
-----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.

下面的實例演示了進程池中的進程如何通信：

import os
import multiprocessing
import time


def write(q):
        print("write啟動(%s)，父進程為(%s)" % (os.getpid(), os.getppid()))
        for i in "python":
                q.put(i)


def read(q):
        print("read啟動(%s)，父進程為(%s)" % (os.getpid(), os.getppid()))
        for i in range(q.qsize()):
                print("read從Queue獲取到消息：%s" % q.get(True))    


if __name__ == "__main__":
        print("(%s) start" % os.getpid())
        q = multiprocessing.Manager().Queue()
        po = multiprocessing.Pool()
        po.apply_async(write, args=(q,))

        time.sleep(2)   
    
        po.apply_async(read, args=(q,))
        po.close()
        po.join()
    
        print("(%s) end" % os.getpid())                                    

運行結果：

(41888) start
write啟動(41894)，父進程為(41888)
read啟動(41895)，父進程為(41888)
read從Queue獲取到消息：p
read從Queue獲取到消息：y
read從Queue獲取到消息：t
read從Queue獲取到消息：h
read從Queue獲取到消息：o
read從Queue獲取到消息：n
(41888) end