Python多進程原理與實現

進程

進程（Process）是計算機中的程序關於某數據集合上的一次運行活動，是系統進行資源分配和調度的基本單位，是操作系統結構的基礎，進程是線程的容器。

進程的概念

1. 進程是一個實體，每一個進程都有自己的內存地址。
   進程一般由程序、數據集、進程控制塊三部分組成。程序用於描述進程要完成的功能，是控制進程執行的指令集；數據集合是程序在執行時所需要的數據和工作區；程序控制塊包含進程的描述信息和控制信息，是進程存在的唯一標志。

2. 進程具有的特征
   動態性：進程是程序的一次執行過程，是臨時的，有生命周期的，是動態生成、動態消亡的；
   並發性：任何進程都可以同其他進程一起並發執行；
   獨立性：進程是系統進行資源分配和調度的一個獨立單位；
   結構性：進程由程序、數據和進程控制塊三部分組成

Python多進程

大部分情況下，想要充分使用多核CPU的資源，就需要在代碼中使用多進程。Python提供了multiprocessing模塊來啟動子進程，並在子進程中執行我們定制的任務。multiprocessing模塊的功能眾多：支持子進程、通信、數據共享和執行不同形式的同步，提供了Process、Queue、Pipe等組件。進程修改的數據僅限於該進程內。

Process類介紹與應用

Process(self, group=None, target=None, name=None, args=(), kwargs={}, daemon=None)

參數說明：

• group參數未使用，值始終為None
• target表示調用的對象，即子進程要執行的任務
• name為子進程的名稱
• args表示調用對象的位置參數元組，必須有逗號，如：args=(1,)
• kwargs表示調用對象的字典，kwargs={'name': 'Tom'}
• daemon表示是否守護進程

下面介紹一下python多進程的開啟和調用

直接調用

import random
from time import sleep
from multiprocessing import Process


def func(name):
    s = random.randint(1, 5)
    print(f'current process is {name}, sleeping {s}s.')
    sleep(s)
    print(f'process {name} is over')


if __name__ == '__main__':
    for i in range(1, 5):
        p = Process(target=func, args=(i,))
        p.start()
    print('main process')

結果如下

main process
current process is 1, sleeping 3s.
current process is 2, sleeping 1s.
current process is 3, sleeping 2s.
current process is 4, sleeping 5s.
process 2 is over
process 3 is over
process 1 is over
process 4 is over

可以看出，各進程之間是並發執行的，先完成任務的進程先結束。

繼承調用

import random
from time import sleep
from multiprocessing import Process


class Func(Process):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        s = random.randint(1, 5)
        print(f'current process is {self.name}, sleeping {s}s.')
        sleep(s)
        print(f'process {self.name} is over')


if __name__ == '__main__':
    for i in range(1, 5):
        p = Func(str(i))
        p.start()
    print('main process')

結果如下

main process
current process is 1, sleeping 3s.
current process is 2, sleeping 1s.
current process is 3, sleeping 4s.
current process is 4, sleeping 3s.
process 2 is over
process 1 is over
process 4 is over
process 3 is over

繼承調用中，start()方法會啟動一個子線程，在該子線程中執行run()方法來實現進程的調用。

Process的join方法

在多進程中，主進程和子進程的執行是同時進行的，如果主進程中有些任務需要等待子進程執行完畢后再執行的話，就需要使用join()方法。

作用：在進程中可以阻塞主進程的執行，直到等待子進程全部完成后，才繼續運行主進程后面的代碼。

我們在直接調用的例子中加入join

import random
from time import sleep
from multiprocessing import Process


def func(name):
    s = random.randint(1, 5)
    print(f'current process is {name}, sleeping {s}s.')
    sleep(s)
    print(f'process {name} is over')


if __name__ == '__main__':
    plist = []
    for i in range(1, 5):
        p = Process(target=func, args=(i,))
        p.start()  # 啟動子進程
        plist.append(p)
    for p in plist:
        p.join()  # 阻塞主進程
    print('main process')
    print('do something')

結果如下

current process is 1, sleeping 5s.
current process is 2, sleeping 4s.
current process is 3, sleeping 5s.
current process is 4, sleeping 4s.
process 2 is over
process 4 is over
process 1 is over
process 3 is over
main process
do something

可以看出，join阻塞了主進程的執行，主進程的打印操作在子進程全部執行完畢后才執行。

守護進程

主進程在開啟子進程的時候，可以將子進程設置為守護進程

守護進程的兩個特點

1. 守護進程會在主進程結束后就會終止（不管守護進程的任務是否執行完畢）
2. 守護進程內無法開啟子進程，否則拋出異常

import random
from time import sleep
from multiprocessing import Process


def func(name):
    s = random.randint(1, 5)
    print(f'current process is {name}, sleeping {s}s.')
    sleep(s)
    print(f'process {name} is over')


if __name__ == '__main__':
    p1 = Process(target=func, args=('daemon',))
    p2 = Process(target=func, args=('sub',))
    p1.daemon = True  # 設置p1為守護進程
    p1.start()  # 啟動守護進程
    p2.start()  # 啟動子進程
    print('main process')
    sleep(2) # 主進程設置兩秒睡眠時間

結果如下

main process
current process is daemon, sleeping 5s.
current process is sub, sleeping 1s.
process sub is over

上面例子中啟動了兩個子進程，其中一個子進程p1設置成了守護進程。從結果我們可以看出，主進程結束后（2秒），守護進程（5秒）也跟隨主進程結束，並沒有打印出后續的內容。

Python進程池

當被操作對象數目不大時，可以直接利用multiprocessing中的Process動態成生多個進程，十幾個還好，但如果是上百個，上千個目標，手動的去限制進程數量卻又太過繁瑣，此時可以發揮進程池的功效。

Pool可以提供指定數量的進程供用戶調用，當有新的請求提交到pool中時，如果池還沒有滿，那么就會創建一個新的進程用來執行該請求；但如果池中的進程數已經達到規定最大值，那么該請求就會等待，直到池中有進程結束，才會創建新的進程來它。

Pool常用參數說明如下

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

請看下面例子

import random
from time import sleep
from multiprocessing import Pool


def func(name):
    s = random.randint(1, 5)
    print(f'current process is {name}, sleeping {s}s.')
    sleep(s)
    print(f'process {name} is over')


if __name__ == '__main__':
    p = Pool(3)
    for i in range(1, 5):
        p.apply_async(func, (i,))
    p.close()
    p.join()
    print('main process')

結果如下

current process is 1, sleeping 1s.
current process is 2, sleeping 5s.
current process is 3, sleeping 4s.
process 1 is over
current process is 4, sleeping 1s.
process 4 is over
process 3 is over
process 2 is over
main process

我們設置進程池的最大值為3，循環的時候先將3個進程放進池子中，等池子中的進程執行結束后（process 1），會將新的進程（process 4）放進池子中來執行。