Python語法學習之程序池與程序鎖詳解

2022-04-11 13:01:37

NICE！大家好，在上一章節，我們學習了 multiprocessing 模組的關於程序的建立與進場常用的方法的相關知識。通過在一個主程序下建立多個子程序可以幫助我們加速程式的執行，並且提高工作效率。不過上一章節文末我們也說過程序的問題，由於每一個程序都會消耗 CPU 與記憶體資源，這樣就不能無限的建立程序的問題，因為會造成記憶體不足或者宕機的情況。

為了解決這個問題我們可以使用多執行緒來替代，或者使用我們今天要學習的內容 —> 程序池。不僅如此，我們在上一章節也說了另一個問題，多程序在同時修改一個檔案的時候可能會存在問題，解決的方法就是給這一個檔案進行上鎖。今天我們就來學習一下程序池與程序鎖，看看它們都能幫助我們怎樣解決問題。

程序池

什麼是程序池

上一章節關於程序的問題我們提到過，程序建立太多的情況下就會對資源消耗過大。為了避免出現這種情況，我們就需要固定程序的數量，這時候就需要程序池的幫助。

我們可以認為程序池就是一個池子，在這個池子裡提前建立好一定數量的程序。見下圖：

比如這個紅色矩形陣列就代表一個程序池子，在這個池子中有6個程序。這6個程序會伴隨程序池一起被建立，不僅如此，我們在學習物件導向的生命週期的時候曾經說過，每個範例化物件在使用完成之後都會被記憶體管家回收。

我們的程序也會伴隨著建立與關閉的過程而被記憶體管家回收，每一個都是如此，建立於關閉程序的過程也會消耗一定的效能。而程序池中的程序當被建立之後就不會被關閉，可以一直被重複使用，從而避免了建立於關閉的資源消耗，也避免了建立於關閉的反覆操作提高了效率。

當然，當我們執行完程式程序池關閉的時候，程序也隨之關閉。

當我們有任務需要被執行的時候，會判斷當前的程序池當中有沒有空閒的程序（所謂空閒的程序其實就是程序池中沒有執行任務的程序）。有程序處於空閒狀態的情況下，任務會找到程序執行該任務。如果當前程序池中的程序都處於非空閒狀態，則任務就會進入等待狀態，直到程序池中有程序處於空閒狀態才會進出程序池從而執行該任務。

這就是程序池的作用。

程序池的建立模組 - multiprocessing

建立程序池函數 - Pool

函數名	介紹	引數	返回值
Pool	程序池的建立	Processcount	程序池物件

Pool功能介紹：通過呼叫 "multiprocessing" 模組的 "Pool" 函數來幫助我們建立 "程序池物件" ，它有一個引數 "Processcount" （一個整數），代表我們這個程序池中建立幾個程序。

程序池的常用方法

當建立了程序池物件之後，我們要對它程序操作，讓我們來看一下都有哪些常用方法(函數)。

函數名	介紹	引數	返回值
apply_async	任務加入程序池(非同步)	func，args	無
close	關閉程序池	無	無
join	等待程序池任務結束	無	無

apply_async 函數：它的功能是將任務加入到程序池中，並且是通過非同步實現的。非同步這個知識我們還沒有學習，先不用關心它到底是什麼意思。它有兩個引數：func 與 agrs , func 是加入程序池中工作的函數；args 是一個元組，代表著籤一個函數的引數，這和我們建立並使用一個程序是完全一致的。
close 函數：當我們使用完程序池之後，通過呼叫 close 函數可以關閉程序池。它沒有任何的引數，也沒有任何的返回值。
join 函數：它和我們上一章節學習的建立程序的 join 函數中方法是一致的。只有程序池中的任務全部執行完畢之後，才會執行後續的任務。不過一般它會伴隨著程序池的關閉（close 函數）才會使用。

apply_async 函數演示案例

接下里我們在 Pycharm 中建立一個指令碼，練習一下關於程序池的使用方法。

定義一個函數，列印輸出該函數每次被執行的次數與該次數的程序號
定義程序池的數量，每一次的執行程序數量最多為該程序池設定的程序數

範例程式碼如下：

# coding:utf-8


import os
import time
import multiprocessing

def work(count):    # 定義一個 work 函數，列印輸出 每次執行的次數 與 該次數的程序號
    print(''work' 函數 第 {} 次執行，程序號為 {}'.format(count, os.getpid()))
    time.sleep(3)
    # print('********')


if __name__ == '__main__':
    pool = multiprocessing.Pool(3)      # 定義程序池的程序數量，同一時間每次執行最多3個程序
    for i in range(21):
        pool.apply_async(func=work, args=(i,))      # 傳入的引數是元組，因為我們只有一個 i 引數，所以我們要寫成 args=(i,)

    time.sleep(15)      # 這裡的休眠時間是必須要加上的，否則我們的程序池還未執行，主程序就已經執行結束，對應的程序池也會關閉。

執行結果如下：

從上圖中我們可以看到每一次都是一次性執行三個程序，每一個程序的程序號是不一樣的，但仔細看會發現存在相同的程序號，這說明程序池的程序號在被重複利用。這證明我們上文介紹的內容，程序池中的程序不會被關閉，可以反覆使用。

而且我們還可以看到每隔3秒都會執行3個程序，原因是我們的程序池中只有3個程序；雖然我們的 for 迴圈中有 21 個任務，work 函數會被執行21次，但是由於我們的程序池中只有3個程序。所以當執行了3個任務之後（休眠3秒），後面的任務等待程序池中的程序處於空閒狀態之後才會繼續執行。

同樣的，程序號在順序上回出現一定的區別，原因是因為我們使用的是一種非同步的方法（非同步即非同步）。這就導致 work 函數一起執行的三個任務會被打亂順序，這也是為什麼我們的程序號出現順序不一致的原因。（更多的非同步知識我們會在非同步的章節進行詳細介紹）

程序池的原理：上述指令碼的案例證實了我們程序池關於程序的限制，只有當我們程序池中的程序處於空閒狀態的時候才會將程序池外等待的任務扔到程序池中工作。

close 函數與 join 函數演示

在上文的指令碼中, 我們使用 time.sleep(15) 幫助我們將主程序阻塞15秒鐘再次退出，所以給了我們程序池足夠的時間完成我們的 work() 函數的迴圈任務。

如果沒有 time.sleep(15) 這句話又怎麼辦呢，其實這裡就可以使用程序的 join 函數了。不過上文我們也提到過，程序的 join() 函數一般都會伴隨程序池的關閉（close 函數）來使用。接下來，我們就將上文指令碼中的 time.sleep(15) 替換成 join() 函數試一下。

範例程式碼如下：

# coding:utf-8


import os
import time
import multiprocessing

def work(count):    # 定義一個 work 函數，列印輸出 每次執行的次數 與 該次數的程序號
    print(''work' 函數 第 {} 次執行，程序號為 {}'.format(count, os.getpid()))
    time.sleep(3)
    # print('********')


if __name__ == '__main__':
    pool = multiprocessing.Pool(3)      # 定義程序池的程序數量，同一時間每次執行最多3個程序
    for i in range(21):
        pool.apply_async(func=work, args=(i,))      # 傳入的引數是元組，因為我們只有一個 i 引數，所以我們要寫成 args=(i,)

    # time.sleep(15) 
    pool.close()
    pool.join()

執行結果如下：

從上面的動圖我們可以看出，work() 函數的任務與程序池中的程序與使用 time.sleep(15)的執行結果一致。

PS：如果我們的主程序會一直執行，不會退出。那麼我們並不需要新增 close() 與 join() 函數，可以讓程序池一直啟動著，直到有任務進來就會執行。

在後面學習 WEB 開發之後，不退出主程序進行工作是家常便飯。還有一些需要長期執行的任務也不會關閉，但要是隻有一次性執行的指令碼，就需要新增 close() 與 join() 函數來保證程序池的任務全部完成之後主程序再退出。當然，如果主程序關閉了，就不會再接受新的任務了，也就代表了程序池的終結。

接下來再看一個例子，在 work 函數中加入一個 return。

這裡大家可能會有一個疑問，在上一章節針對程序的知識點明明說的是程序無法獲取返回值，那麼這裡的 work() 函數增加的 return 又有什麼意義呢？

其實不然，在我們的使用程序池的 apply_async 方法時，是通過非同步的方式實現的，而非同步是可以獲取返回值的。針對上述指令碼，我們在 for迴圈中針對每一個非同步 apply_async 新增一個變數名，從而獲取返回值。

範例程式碼如下：

# coding:utf-8


import os
import time
import multiprocessing

def work(count):    # 定義一個 work 函數，列印輸出 每次執行的次數 與 該次數的程序號
    print(''work' 函數 第 {} 次執行，程序號為 {}'.format(count, os.getpid()))
    time.sleep(3)
    return ''work' 函數 result 返回值為：{}, 程序ID為：{}'.format(count, os.getpid())


if __name__ == '__main__':
    pool = multiprocessing.Pool(3)      # 定義程序池的程序數量，同一時間每次執行最多3個程序
    results = []
    for i in range(21):
        result = pool.apply_async(func=work, args=(i,))      # 傳入的引數是元組，因為我們只有一個 i 引數，所以我們要寫成 args=(i,)
        results.append(result)

    for result in results:
        print(result.get())     # 可以通過這個方式返回 apply_async 的返回值，
                                # 通過這種方式也不再需要 使用 close()、join() 函數就可以正常執行。

    # time.sleep(15)      # 這裡的休眠時間是必須要加上的，否則我們的程序池還未執行，主程序就已經執行結束，對應的程序池也會關閉。
    # pool.close()
    # pool.join()

執行結果如下：

從執行結果可以看出，首先 work() 函數被執行緒池的執行緒執行了一遍，當第一組任務執行完畢緊接著執行第二次執行緒池任務的時候，列印輸出了 apply_async 的返回值，證明返回值被成功的返回了。然後繼續下一組的任務…

這些都是主要依賴於非同步，關於非同步的更多知識會在非同步的章節進行詳細的介紹。

程序鎖

程序鎖的概念

鎖：大家都知道，我們可以給一個大門上鎖。

結合這個場景來舉一個例子：比如現在有多個程序同時衝向一個 "大門" ，當前門內是沒有 "人"的(其實就是程序)，鎖也沒有鎖上。當有一個程序進去之後並且把 “門” 鎖上了，這時候門外的那些程序是進不來的。在門內的 “人” ，可以在 “門” 內做任何事情且不會被幹擾。當它出來之後，會解開門鎖。這時候又有一個 “人” 進去了門內，並且重複這樣的操作，這就是程序鎖。它可以讓鎖後面的工作只能被一個任務來處理，只有它解鎖之後下一個任務才會進入，這就是 “鎖” 的概念。

而程序鎖就是僅針對於程序有效的鎖，當程序的任務開始之後，就會被上一把 “鎖”；與之對應的是執行緒鎖，它們的原理幾乎是一樣的。

程序鎖的加鎖與解鎖

程序鎖的使用方法：

通過 multiprocessing 匯入 Manager 類

from multiprocessing import Manager

然後範例化 Manager

manager = Manager()

再然後通過範例化後的 manager 呼叫它的 Lock() 函數

lock = manager.Lock()

接下來，就需要操作這個 lock 物件的函數

函數名	介紹	引數	返回值
acquire	上鎖	無	無
release	解鎖（開鎖）	無	無

程式碼範例如下：

# coding:utf-8


import os
import time
import multiprocessing


def work(count, lock):    # 定義一個 work 函數，列印輸出 每次執行的次數 與 該次數的程序號,增加執行緒鎖。
    lock.acquire()        # 上鎖
    print(''work' 函數 第 {} 次執行，程序號為 {}'.format(count, os.getpid()))
    time.sleep(3)
    lock.release()        # 解鎖
    return ''work' 函數 result 返回值為：{}, 程序ID為：{}'.format(count, os.getpid())


if __name__ == '__main__':
    pool = multiprocessing.Pool(3)      # 定義程序池的程序數量，同一時間每次執行最多3個程序
    manager = multiprocessing.Manager()
    lock = manager.Lock()
    results = []
    for i in range(21):
        result = pool.apply_async(func=work, args=(i, lock))      # 傳入的引數是元組，因為我們只有一個 i 引數，所以我們要寫成 args=(i,)
        # results.append(result)


    # time.sleep(15)      # 這裡的休眠時間是必須要加上的，否則我們的程序池還未執行，主程序就已經執行結束，對應的程序池也會關閉。
    pool.close()
    pool.join()

執行結果如下：