Python編程進(jìn)階，輕松掌握多線程和多進(jìn)程
1、簡(jiǎn)介
我們將討論如何利用Python執(zhí)行多線程和多進(jìn)程任務(wù)。它們提供了在單個(gè)進(jìn)程或多個(gè)進(jìn)程之間執(zhí)行并發(fā)操作的方法。并行和并發(fā)執(zhí)行可以提高系統(tǒng)的速度和效率。
在討論多線程和多進(jìn)程的基礎(chǔ)知識(shí)之后，我們還將討論使用Python庫(kù)實(shí)現(xiàn)它們的實(shí)際方法。
首先簡(jiǎn)要討論并行系統(tǒng)的好處。
- 改進(jìn)的性能：有了并發(fā)執(zhí)行任務(wù)的能力，可以減少執(zhí)行時(shí)間并提高系統(tǒng)的整體性能。
- 可擴(kuò)展性：可以將一個(gè)大任務(wù)分解為多個(gè)較小的子任務(wù)，并為它們分配獨(dú)立的核心或線程，讓它們獨(dú)立執(zhí)行。這在大規(guī)模系統(tǒng)中非常有用。
- 高效的I/O操作：通過(guò)并發(fā)的幫助，CPU不必等待進(jìn)程完成其I/O操作。CPU可以立即開(kāi)始執(zhí)行下一個(gè)進(jìn)程，直到前一個(gè)進(jìn)程忙于其I/O操作。
- 資源優(yōu)化：通過(guò)分割資源，可以防止單個(gè)進(jìn)程占用所有資源。這可以避免較小進(jìn)程的Starvation（饑餓）問(wèn)題。
并行計(jì)算的優(yōu)勢(shì)
以上是需要并發(fā)或并行執(zhí)行的一些常見(jiàn)原因。現(xiàn)在，回到主題，即多線程和多進(jìn)程，并討論它們的主要區(qū)別。
2、什么是多線程？
多線程是在單個(gè)進(jìn)程中實(shí)現(xiàn)并行性的一種方法，能夠執(zhí)行同時(shí)進(jìn)行的任務(wù)。在單個(gè)進(jìn)程內(nèi)可以創(chuàng)建多個(gè)線程，并在該進(jìn)程內(nèi)并行執(zhí)行較小的任務(wù)。
單個(gè)進(jìn)程中的線程共享一個(gè)公共內(nèi)存空間，但它們的堆棧跟蹤和寄存器是獨(dú)立的。由于共享內(nèi)存，它們的計(jì)算成本較低。
單線程和多線程Env.
Python中的多線程主要用于執(zhí)行I/O操作，即如果程序的某個(gè)部分正在執(zhí)行I/O操作，則其余程序可以保持響應(yīng)。然而，在Python的實(shí)現(xiàn)中，由于全局解釋器鎖（GIL）的存在，多線程無(wú)法實(shí)現(xiàn)真正的并行性。
簡(jiǎn)而言之，GIL是一個(gè)互斥鎖，一次只允許一個(gè)線程與Python字節(jié)碼交互，即使在多線程模式下，一次也只能有一個(gè)線程執(zhí)行字節(jié)碼。
這樣做是為了在CPython中保持線程安全，但它限制了多線程的性能優(yōu)勢(shì)。為了解決這個(gè)問(wèn)題，Python有一個(gè)單獨(dú)的多進(jìn)程庫(kù)，我們將在之后進(jìn)行討論。
什么是守護(hù)線程？
不斷在后臺(tái)運(yùn)行的線程稱為守護(hù)線程。它們的主要工作是支持主線程或非守護(hù)線程。守護(hù)線程不會(huì)阻塞主線程的執(zhí)行，甚至?xí)谥骶€程執(zhí)行完畢后繼續(xù)運(yùn)行。
在Python中，守護(hù)線程主要用作垃圾回收器。它會(huì)默認(rèn)銷毀所有無(wú)用的對(duì)象并釋放內(nèi)存，以便主線程可以正常使用和執(zhí)行。
3、什么是多進(jìn)程？
多進(jìn)程用于執(zhí)行多個(gè)進(jìn)程的并行執(zhí)行。它可以幫助實(shí)現(xiàn)真正的并行性，因?yàn)榭梢酝瑫r(shí)執(zhí)行不同的進(jìn)程，并且每個(gè)進(jìn)程都擁有自己的內(nèi)存空間。它使用CPU的獨(dú)立核心，并且在執(zhí)行進(jìn)程間的數(shù)據(jù)交換時(shí)也很有幫助。
與多線程相比，多進(jìn)程的計(jì)算成本更高，因?yàn)椴皇褂霉蚕韮?nèi)存空間。不過(guò)，它允許進(jìn)行獨(dú)立執(zhí)行，并克服了全局解釋器鎖的限制。
多進(jìn)程環(huán)境
上圖展示了一個(gè)多進(jìn)程環(huán)境，在該環(huán)境中，一個(gè)主進(jìn)程創(chuàng)建了兩個(gè)獨(dú)立的進(jìn)程，并為它們分配了不同的工作。
4、多線程實(shí)現(xiàn)
現(xiàn)在，我們使用Python實(shí)現(xiàn)一個(gè)基本的多線程示例。Python有一個(gè)內(nèi)置的threading模塊用于多線程實(shí)現(xiàn)。
- 導(dǎo)入庫(kù)：
import threading
import os- 計(jì)算平方的函數(shù)：
這是一個(gè)用于計(jì)算數(shù)字平方的簡(jiǎn)單函數(shù)，它接受一個(gè)數(shù)字列表作為輸入，并輸出列表中每個(gè)數(shù)字的平方，同時(shí)輸出使用的線程名稱和與該線程關(guān)聯(lián)的進(jìn)程ID。
def calculate_squares(numbers):
for num in numbers:
square = num * num
print(
f"Square of the number {num} is {square} | Thread Name {threading.current_thread().name} | PID of the process {os.getpid()}"
)- 主函數(shù)：
本示例有一個(gè)數(shù)字列表，將其平均分成兩半，并分別命名為first_half和second_half。現(xiàn)在，將為這些列表分配兩個(gè)獨(dú)立的線程t1和t2。
Thread函數(shù)創(chuàng)建一個(gè)新線程，該線程接受一個(gè)帶有參數(shù)列表的函數(shù)作為輸入。還可以為線程分配一個(gè)單獨(dú)的名稱。
.start()函數(shù)將開(kāi)始執(zhí)行這些線程，而.join()函數(shù)將阻塞主線程的執(zhí)行，直到給定的線程完全執(zhí)行完畢。
if __name__ == "__main__":
numbers = [1, 2, 3, 4, 5, 6, 7, 8]
half = len(numbers) // 2
first_half = numbers[:half]
second_half = numbers[half:]
t1 = threading.Thread(target=calculate_squares, name="t1", args=(first_half,))
t2 = threading.Thread(target=calculate_squares, name="t2", args=(second_half,))
t1.start()
t2.start()
t1.join()
t2.join()輸出：
Square of the number 1 is 1 | Thread Name t1 | PID of the process 345
Square of the number 2 is 4 | Thread Name t1 | PID of the process 345
Square of the number 5 is 25 | Thread Name t2 | PID of the process 345
Square of the number 3 is 9 | Thread Name t1 | PID of the process 345
Square of the number 6 is 36 | Thread Name t2 | PID of the process 345
Square of the number 4 is 16 | Thread Name t1 | PID of the process 345
Square of the number 7 is 49 | Thread Name t2 | PID of the process 345
Square of the number 8 is 64 | Thread Name t2 | PID of the process 345注意：上述創(chuàng)建的所有線程都是非守護(hù)線程。要?jiǎng)?chuàng)建守護(hù)線程，需要編寫t1.setDaemon(True)，將線程t1設(shè)置為守護(hù)線程。
現(xiàn)在來(lái)了解一下上述代碼生成的輸出結(jié)果。可以觀察到兩個(gè)線程的進(jìn)程ID（即PID）保持不變，這意味著這兩個(gè)線程屬于同一個(gè)進(jìn)程。
還可以觀察到輸出并非按順序生成。第一行中可以看到是線程1生成的輸出，然后在第三行是線程2生成的輸出，接著在第四行，再次是線程1生成的輸出。這清楚地表明這些線程是同時(shí)工作的。
并發(fā)并不意味著這兩個(gè)線程并行執(zhí)行，因?yàn)橐淮沃挥幸粋€(gè)線程被執(zhí)行。它不會(huì)減少執(zhí)行時(shí)間，與順序執(zhí)行所需的時(shí)間相同。CPU開(kāi)始執(zhí)行一個(gè)線程，但在中途離開(kāi)，并切換到另一個(gè)線程，過(guò)一段時(shí)間后，又回到主線程，并從上次離開(kāi)的地方開(kāi)始執(zhí)行。
5、多進(jìn)程實(shí)現(xiàn)
目前對(duì)多線程及其實(shí)現(xiàn)方式和限制已經(jīng)有基本的了解。現(xiàn)在，是時(shí)候?qū)W習(xí)多進(jìn)程的實(shí)現(xiàn)以及如何克服這些限制了。
在這里將沿用相同的示例，但不再創(chuàng)建兩個(gè)獨(dú)立的線程，而是創(chuàng)建兩個(gè)獨(dú)立的進(jìn)程，并討論觀察結(jié)果。
- 導(dǎo)入庫(kù)：
from multiprocessing import Process
import os本例將使用multiprocessing模塊來(lái)創(chuàng)建獨(dú)立的進(jìn)程。
- 計(jì)算平方的函數(shù)：
該函數(shù)將保持不變。只是在這里刪除了有關(guān)線程信息的打印語(yǔ)句。
def calculate_squares(numbers):
for num in numbers:
square = num * num
print(
f"Square of the number {num} is {square} | PID of the process {os.getpid()}"
)- 主函數(shù)：
主函數(shù)有一些修改。只是創(chuàng)建了一個(gè)獨(dú)立的進(jìn)程，而不是線程。
if __name__ == "__main__":
numbers = [1, 2, 3, 4, 5, 6, 7, 8]
half = len(numbers) // 2
first_half = numbers[:half]
second_half = numbers[half:]
p1 = Process(target=calculate_squares, args=(first_half,))
p2 = Process(target=calculate_squares, args=(second_half,))
p1.start()
p2.start()
p1.join()
p2.join()輸出：
Square of the number 1 is 1 | PID of the process 1125
Square of the number 2 is 4 | PID of the process 1125
Square of the number 3 is 9 | PID of the process 1125
Square of the number 4 is 16 | PID of the process 1125
Square of the number 5 is 25 | PID of the process 1126
Square of the number 6 is 36 | PID of the process 1126
Square of the number 7 is 49 | PID of the process 1126
Square of the number 8 is 64 | PID of the process 1126可以觀察到，每個(gè)列表都由一個(gè)獨(dú)立的進(jìn)程執(zhí)行。它們具有不同的進(jìn)程ID。為了檢查進(jìn)程是否已并行執(zhí)行，需要?jiǎng)?chuàng)建一個(gè)單獨(dú)的環(huán)境，下面我們將討論這一點(diǎn)。
計(jì)算是否使用多進(jìn)程的運(yùn)行時(shí)間
為了檢查是否獲得了真正的并行性，在這里將計(jì)算使用和不使用多進(jìn)程的算法運(yùn)行時(shí)間。
為此，需要一個(gè)包含超過(guò)10^6個(gè)整數(shù)的大型整數(shù)列表。可以使用random庫(kù)生成一個(gè)列表。此處將使用Python的time模塊來(lái)計(jì)算運(yùn)行時(shí)間。下面是實(shí)現(xiàn)的代碼，代碼本身很容易理解，也可以隨時(shí)查看代碼注釋。
from multiprocessing import Process
import os
import time
import random
def calculate_squares(numbers):
for num in numbers:
square = num * num
if __name__ == "__main__":
numbers = [
random.randrange(1, 50, 1) for i in range(10000000)
] # 創(chuàng)建一個(gè)包含10^7個(gè)整數(shù)的隨機(jī)列表。
half = len(numbers) // 2
first_half = numbers[:half]
second_half = numbers[half:]
# ----------------- 創(chuàng)建單進(jìn)程環(huán)境 ------------------------#
start_time = time.time() # 開(kāi)始計(jì)時(shí)（不使用多進(jìn)程）
p1 = Process(
target=calculate_squares, args=(numbers,)
) # 單進(jìn)程P1執(zhí)行整個(gè)列表
p1.start()
p1.join()
end_time = time.time() # 結(jié)束計(jì)時(shí)（不使用多進(jìn)程）
print(f"Execution Time Without Multiprocessing: {(end_time-start_time)*10**3}ms")
# ----------------- 創(chuàng)建多進(jìn)程環(huán)境 ------------------------#
start_time = time.time() # 開(kāi)始計(jì)時(shí)（使用多進(jìn)程）
p2 = Process(target=calculate_squares, args=(first_half,))
p3 = Process(target=calculate_squares, args=(second_half,))
p2.start()
p3.start()
p2.join()
p3.join()
end_time = time.time() # 結(jié)束計(jì)時(shí)（使用多進(jìn)程）
print(f"Execution Time With Multiprocessing: {(end_time-start_time)*10**3}ms")輸出：
Execution Time Without Multiprocessing: 619.8039054870605ms
Execution Time With Multiprocessing: 321.70287895202637ms可以觀察到，使用多進(jìn)程的時(shí)間幾乎是不使用多進(jìn)程時(shí)間的一半。這表明這兩個(gè)進(jìn)程在同一時(shí)間內(nèi)并行執(zhí)行，并展示了真正的并行性行為。
