Python編程之多進(jìn)程(multiprocessing)詳解

　　Python中的multiprocessing模塊提供了一種創(chuàng)建和管理進(jìn)程的方式，使得可以利用多個CPU來加速程序運(yùn)行。在這里，我會詳細(xì)介紹Python中的多進(jìn)程編程，包括以下內(nèi)容：

　　一.多進(jìn)程概念

　　二.multiprocessing模塊介紹

　　三.進(jìn)程池

　　四.進(jìn)程通信

　　五.multiprocessing的一些注意事項

　　一.多進(jìn)程概念

　　多進(jìn)程是指在同一時間內(nèi)，同時執(zhí)行多個程序或多個部分的程序。每個進(jìn)程都擁有自己的地址空間、內(nèi)存、文件描述符和其他系統(tǒng)資源。多進(jìn)程的好處在于可以使程序并行執(zhí)行，從而提高程序的運(yùn)行效率。

　　二.multiprocessing模塊介紹

　　multiprocessing模塊提供了一個Process類，可以用來創(chuàng)建和管理進(jìn)程。下面是一個簡單的示例：

import multiprocessing

def worker():
    """該函數(shù)將在子進(jìn)程中執(zhí)行"""
    print('Worker')

if __name__ == '__main__':
    # 創(chuàng)建子進(jìn)程
    p = multiprocessing.Process(target=worker)
    # 啟動子進(jìn)程
    p.start()
    # 等待子進(jìn)程結(jié)束
    p.join()

　　在上面的代碼中，worker函數(shù)將在子進(jìn)程中執(zhí)行。首先，創(chuàng)建了一個Process對象，指定target參數(shù)為worker函數(shù)。然后，通過調(diào)用start方法啟動子進(jìn)程，最后調(diào)用join方法等待子進(jìn)程結(jié)束。

　　三. 進(jìn)程池

　　如果需要創(chuàng)建大量的進(jìn)程，那么使用Process類可能會導(dǎo)致系統(tǒng)資源的浪費(fèi)。此時，可以使用Pool類來創(chuàng)建進(jìn)程池。下面是一個簡單的示例：

import multiprocessing

def worker(num):
    """該函數(shù)將在子進(jìn)程中執(zhí)行"""
    print('Worker %d' % num)

if __name__ == '__main__':
    # 創(chuàng)建進(jìn)程池
    pool = multiprocessing.Pool(4)
    # 啟動進(jìn)程池中的進(jìn)程
    pool.map(worker, range(10))
    # 關(guān)閉進(jìn)程池
    pool.close()
    # 等待進(jìn)程池中的進(jìn)程結(jié)束
    pool.join()

　　在上面的代碼中，Pool類的構(gòu)造函數(shù)中指定了進(jìn)程池的大小為4，然后通過調(diào)用map方法來啟動進(jìn)程池中的進(jìn)程。map方法會將worker函數(shù)和range(10)序列中的每個元素一一對應(yīng)，然后將它們作為參數(shù)傳遞給進(jìn)程池中的進(jìn)程。最后，調(diào)用close方法關(guān)閉進(jìn)程池，并調(diào)用join方法等待所有進(jìn)程結(jié)束。

　　四. 進(jìn)程通信

　　在多進(jìn)程編程中，不同的進(jìn)程之間需要進(jìn)行通信。multiprocessing模塊提供了多種進(jìn)程間通信的方式，例如使用隊列、管道、共享內(nèi)存等。

　　(1)隊列

　　隊列是一種常用的進(jìn)程間通信方式。multiprocessing模塊中提供了Queue類，可以用來創(chuàng)建隊列。下面是一個簡單的示例：

import multiprocessing

def producer(q):
    """該函數(shù)將在生產(chǎn)者進(jìn)程中執(zhí)行"""
    for i in range(10):
        q.put(i)

def consumer(q):
    """該函數(shù)將在消費(fèi)者進(jìn)程中執(zhí)行"""
    while True:
        item = q.get()
        if item is None:
            break
        print(item)

if __name__ == '__main__':
    # 創(chuàng)建隊列
    q = multiprocessing.Queue()
    # 創(chuàng)建生產(chǎn)者進(jìn)程
    p1 = multiprocessing.Process(target=producer, args=(q,))
    # 創(chuàng)建消費(fèi)者進(jìn)程
    p2 = multiprocessing.Process(target=consumer, args=(q,))
    # 啟動進(jìn)程
    p1.start()
    p2.start()
    # 等待進(jìn)程結(jié)束
    p1.join()
    # 發(fā)送結(jié)束信號
    q.put(None)
    p2.join()

　　在上面的代碼中，首先創(chuàng)建了一個Queue對象，然后創(chuàng)建了一個生產(chǎn)者進(jìn)程和一個消費(fèi)者進(jìn)程。生產(chǎn)者進(jìn)程通過調(diào)用put方法將0~9的數(shù)字放入隊列中，消費(fèi)者進(jìn)程通過調(diào)用get方法從隊列中獲取數(shù)據(jù)，并將其打印出來。最后，調(diào)用put方法發(fā)送結(jié)束信號，然后等待兩個進(jìn)程結(jié)束。

　　(2)管道

　　管道是另一種常用的進(jìn)程間通信方式。multiprocessing模塊中提供了Pipe類，可以用來創(chuàng)建管道。下面是一個簡單的示例：

import multiprocessing

def producer(conn):
    """該函數(shù)將在生產(chǎn)者進(jìn)程中執(zhí)行"""
    for i in range(10):
        conn.send(i)
    conn.close()

def consumer(conn):
    """該函數(shù)將在消費(fèi)者進(jìn)程中執(zhí)行"""
    while True:
        item = conn.recv()
        if item is None:
            break
        print(item)

if __name__ == '__main__':
    # 創(chuàng)建管道
    conn1, conn2 = multiprocessing.Pipe()
    # 創(chuàng)建生產(chǎn)者進(jìn)程
    p1 = multiprocessing.Process(target=producer, args=(conn1,))
    # 創(chuàng)建消費(fèi)者進(jìn)程
    p2 = multiprocessing.Process(target=consumer, args=(conn2,))
    # 啟動進(jìn)程
    p1.start()
    p2.start()
    # 等待進(jìn)程結(jié)束
    p1.join()
    # 發(fā)送結(jié)束信號
    conn1.send(None)
    p2.join()

　　在上面的代碼中，首先創(chuàng)建了一個管道，然后創(chuàng)建了一個生產(chǎn)者進(jìn)程和一個消費(fèi)者進(jìn)程。生產(chǎn)者進(jìn)程通過調(diào)用send方法將0~9的數(shù)字發(fā)送到管道中，消費(fèi)者進(jìn)程通過調(diào)用recv方法從管道中獲取數(shù)據(jù)，并將其打印出來。最后，調(diào)用send方法發(fā)送結(jié)束信號，然后等待兩個進(jìn)程結(jié)束。

　　(3)共享內(nèi)存

　　共享內(nèi)存是一種高效的進(jìn)程間通信方式，它允許多個進(jìn)程共享同一塊內(nèi)存區(qū)域。multiprocessing模塊中提供了Value和Array類，可以用來創(chuàng)建共享內(nèi)存。下面是一個簡單的示例：

import multiprocessing

def worker1(n):
    """該函數(shù)將在進(jìn)程1中執(zhí)行"""
    n.value += 1
    print('worker1:', n.value)

def worker2(n):
    """該函數(shù)將在進(jìn)程2中執(zhí)行"""
    n.value += 1
    print('worker2:', n.value)

if __name__ == '__main__':
    # 創(chuàng)建共享內(nèi)存
    n = multiprocessing.Value('i', 0)
    # 創(chuàng)建進(jìn)程1
    p1 = multiprocessing.Process(target=worker1, args=(n,))
    # 創(chuàng)建進(jìn)程2
    p2 = multiprocessing.Process(target=worker2, args=(n,))
    # 啟動進(jìn)程
    p1.start()
    p2.start()
    # 等待進(jìn)程結(jié)束
    p1.join()
    p2.join()

　　在上面的代碼中，首先創(chuàng)建了一個Value對象，用于存儲一個整數(shù)值。然后創(chuàng)建了兩個進(jìn)程，每個進(jìn)程都會將共享內(nèi)存中的值加1，并將其打印出來。最后，等待兩個進(jìn)程結(jié)束。

　　除了Value類之外，multiprocessing模塊還提供了Array類，用于創(chuàng)建共享內(nèi)存數(shù)組。下面是一個簡單的示例：

　　在上面的代碼中，首先創(chuàng)建了一個Array對象，用于存儲一個整數(shù)數(shù)組。然后創(chuàng)建了兩個進(jìn)程，每個進(jìn)程都會將共享內(nèi)存數(shù)組中的第一個元素加1，并將其打印出來。最后，等待兩個進(jìn)程結(jié)束。

　　五.multiprocessing的一些注意事項

　　在使用 multiprocessing 模塊進(jìn)行多進(jìn)程編程時，需要注意以下幾點(diǎn)：

　　1.全局變量的共享問題

　　每個子進(jìn)程都有自己的內(nèi)存空間，因此全局變量在多進(jìn)程之間不能直接共享。如果需要共享數(shù)據(jù)，可以使用 multiprocessing.Value 或 multiprocessing.Array 來創(chuàng)建共享內(nèi)存。

　　2.進(jìn)程間通信問題

　　多個進(jìn)程之間需要相互通信，可以使用 multiprocessing.Queue 或 multiprocessing.Pipe 來進(jìn)行進(jìn)程間通信。

　　3.進(jìn)程池的使用

　　如果需要同時啟動多個進(jìn)程，可以使用進(jìn)程池來管理進(jìn)程。進(jìn)程池可以避免頻繁地啟動和關(guān)閉進(jìn)程所帶來的開銷，提高程序的效率。

　　4.子進(jìn)程的異常處理

　　每個子進(jìn)程都是一個獨(dú)立的進(jìn)程，當(dāng)子進(jìn)程出現(xiàn)異常時，主進(jìn)程并不會收到通知。因此需要在子進(jìn)程中進(jìn)行異常處理，并將異常信息通過進(jìn)程間通信的方式傳遞給主進(jìn)程。

　　5.進(jìn)程的啟動方式

　　可以使用 multiprocessing.Process 來創(chuàng)建進(jìn)程，也可以使用 multiprocessing.Pool 來創(chuàng)建進(jìn)程池。進(jìn)程池可以方便地管理多個進(jìn)程，避免手動啟動和關(guān)閉進(jìn)程所帶來的麻煩。