您现在的位置：程式師世界 >> 編程語言 > >> 更多編程語言 >> Python

Python高級教程筆記：多線程初步

編輯：Python

0x01 線程、進程初步

進程是系統分配資源的最小單位。一個程序至少有一個進程。

線程是OS中可以進行調度的最小單位，被包含在進程之中，是進程中的實際運作單位。

一個進程中可以存在多個線程，在單核CPU中每個進程中同時刻只能運行一個線程，只有在多核CPU中才能存在線程並發的情況。

進程擁有自己的獨立地址空間（相對地址空間），內存、數據棧等，因此進程占用資源較多。由於進程的資源獨立，所以通訊不方便，只能使用進程間通訊（IPC）。但是，同一個進程中的線程都有共性多個線程共享同一個進程的虛擬空間。線程共享的環境包括進程代碼段、進程的公有數據及文件等，利用這些共享的數據，線程之間很容易實現通信。

此外，操作系統在創建進程時，必須為該進程分配獨立的內存空間，並分配大量的相關資源，但創建線程則簡單得多。

因此，使用多線程來實現並發比使用多進程的性能要高得多。

0x02 線程實現

普通的創建方式：需要繼承threading.Thread，繼承出的子類需要實現run方法。

import threading
import time
class Mythread(threading.Thread):
def __init__(self, n):
super(Mythread, self).__init__()
self.n = n
def run(self):
print("task", self.n)
time.sleep(1)
print('2s')
time.sleep(1)
print('1s')
time.sleep(1)
print('0s')
time.sleep(1)
print("finish", self.n)
if __name__ == "__main__":
t1 = Mythread("t1")
t1.start()
t2 = Mythread("t2")
t3 = Mythread("t3")
t2.start()
t3.start()
t4 = Mythread("t4")
t5 = Mythread("t5")
t4.start()
t5.start()

運行結果如下：

task t1
task t2
tasktask t4
t3
task t5
2s
2s
2s2s
2s
1s
1s
1s1s
1s
0s
0s
0s0s
0s
finish t1
finish t2
finishfinish t3
t4
finish t5

之所以出現排版這麼奇葩的情況，就是因為各個線程只會運行一定量的時間片。一個線程如果運行時間到了之後，CPU就會轉而執行其他的線程。

比如線程 i 執行一個循環，執行了一半之後cpu暫停執行線程 i 了，轉而執行線程 j …

0x03 守護線程

若在主線程中創建了子線程，當主線程結束時根據子線程daemon（設置Thread.setDaemon(True)）屬性值的不同可能會發生下面的兩種情況之一：

如果某個子線程的daemon屬性為True，主線程運行結束時不對這個子線程進行檢查而直接退出，同時所有daemon值為True的子線程將隨主線程一起結束，而不論是否運行完成。
如果某個子線程的daemon屬性為False，主線程結束時會檢測該子線程是否結束，如果該子線程還在運行，則主線程會等待它完成後再退出。

實驗代碼如下：

from threading import Thread, Lock
import os, time
def work():
global n
#lock.acquire()
temp = n
time.sleep(.1)
n = temp - 1
print(n)
#lock.release()
if __name__ == "__main__":
lock = Lock()
n = 100
for i in range(10):
p = Thread(target=work)
p.setDaemon(True)
p.start()

結果如下：

可以看見，主線程在執行完之後直退出了，並沒有等待子線程結束完才退出。

0x04 多線程共享全局變量

在同一個進程中的多線程是共享資源的，比如變量、文件等等。

import threading
import time
g_num = 100.0
def work1():
global g_num
for i in range(5):
g_num += 1
time.sleep(2)
print(f"work 1 :g_num is {
g_num}")
def work2():
global g_num
for i in range(10):
g_num -= .5
time.sleep(1)
print(f"work 2 :g_num is {
g_num}")
if __name__ == "__main__":
t1 = threading.Thread(target=work1)
t2 = threading.Thread(target=work2)
t1.start()
t2.start()

運行結果如下：

work 2 :g_num is 100.5
work 1 :g_num is 100.0
work 2 :g_num is 101.0
work 2 :g_num is 100.5
work 1 :g_num is 100.0
work 2 :g_num is 101.0
work 2 :g_num is 100.5
work 1 :g_num is 100.0
work 2 :g_num is 101.0
work 2 :g_num is 100.5
work 1 :g_num is 100.0
work 2 :g_num is 101.0
work 2 :g_num is 100.5
work 1 :g_num is 100.0
work 2 :g_num is 100.0

0x05 互斥鎖

為了解決線程不同步的問題，對於有些資源（比如變量、代碼）我們需要確保一件事：在任意時刻只有0個或者1個線程可以訪問資源。解決辦法就是使用互斥鎖Lock。

from threading import Thread, Lock
import os, time
def work():
global n
lock.acquire()
temp = n
time.sleep(.1)
n = temp - 1
print(n)
lock.release()
if __name__ == "__main__":
lock = Lock()
n = 100
l = []
for i in range(10):
p = Thread(target=work)
l.append(p)
p.start()
# 這裡的join是指讓主線程等待新開的線程運行結束之後再結束
for p in l:
p.join()

加入線程同步之後的運行結果：

from threading import Thread, Lock
import os, time
def work():
global n
#lock.acquire()
temp = n
time.sleep(.1)
n = temp - 1
print(n)
#lock.release()
if __name__ == "__main__":
lock = Lock()
n = 100
l = []
for i in range(10):
p = Thread(target=work)
l.append(p)
p.start()
# 這裡的join是指讓主線程等待新開的線程運行結束之後再結束
for p in l:
p.join()

沒有線程同步的運行結果：

0x06 RLock鎖（可重入鎖，遞歸鎖）

如果用了Lock鎖，那麼下面的代碼會發生阻塞：

import threading
lock = threading.Lock()
lock.acquire()
for i in range(10):
print('獲取第二把鎖')
lock.acquire()
print(f'test.......{
i}')
lock.release()
lock.release()

但是這個鎖換成RLock鎖，就不會發生阻塞。

RLock鎖會檢查申請使用鎖的是哪個線程。在鎖被占用的情況下，

如果申請使用鎖的是同一個線程，那麼RLock會為該線程維護一個計數器，並且給計數器+1；
如果申請使用鎖的線程不是同一個，則會被阻塞。

0x07 信號量（Semaphore）

互斥鎖只允許一個線程更改數據，但是信號量允許一定數量的線程更改數據。

比如只有三個蹲坑的撤碩。

import threading
import time
def run(n, semaphore):
semaphore.acquire() #加鎖
time.sleep(1)
print("run the thread:%s\n" % n)
semaphore.release() #釋放
if __name__ == '__main__':
num = 0
semaphore = threading.BoundedSemaphore(5) # 最多允許5個線程同時運行
for i in range(22):
t = threading.Thread(target=run, args=("t-%s" % i, semaphore))
t.start()
while threading.active_count() != 1:
pass # print threading.active_count()
else:
print('-----all threads done-----')

0x08 事件類（Event）

Event類用於主線程控制其它線程的運行，事件是一個簡單的線程同步對象，其主要提供以下幾個方法：

clear 將flag設置為“False”
set 將flag設置為“True”
is_set 判斷是否設置了flag
wait 會一直監聽flag，如果沒有檢測到flag就一直處於阻塞狀態

事件處理的機制：全局定義了一個“flag”，當flag值為“False”，那麼event.wait()就會阻塞，當flag值為“True”，那麼event.wait()便不再阻塞。

#利用Event類模擬紅綠燈
import threading
import time
event = threading.Event()
def lighter():
count = 0
event.set() #初始值為綠燈
while True:
if 5 < count <=10 :
event.clear() # 紅燈，清除標志位
print("\33[41;1mred light is on...\033[0m")
elif count > 10:
event.set() # 綠燈，設置標志位
count = 0
else:
print("\33[42;1mgreen light is on...\033[0m")
time.sleep(1)
count += 1
def car(name):
while True:
if event.is_set(): #判斷是否設置了標志位
print("[%s] running..."%name)
time.sleep(1)
else:
print("[%s] sees red light,waiting..."%name)
event.wait()
print("[%s] green light is on,start going..."%name)
light = threading.Thread(target=lighter,)
light.start()
car = threading.Thread(target=car,args=("MINI",))
car.start()

0x09 全局解釋器鎖(GIL)

在非python環境中，單核情況下，同時只能有一個任務執行。多核時可以支持多個線程同時執行。

但在python中，無論有多少核，同時只能執行一個線程。究其原因，就是GIL的存在。

這也就是人們說Python是偽多線程的原因。

GIL全稱全局解釋器鎖，來源是python設計之初的考慮，為了數據安全所做的決定。

某個線程想要執行，必須先拿到GIL，我們可以把GIL看作是“通行證”，並且在一個python進程中，GIL只有一個。拿不到通行證的線程，就不允許進入CPU執行。GIL只在cpython中才有，因為cpython調用的是c語言的原生線程，所以他不能直接操作cpu，只能利用GIL保證同一時間只能有一個線程拿到數據。

而在pypy和jpython中是沒有GIL的。

Python多線程的工作過程：

python在使用多線程的時候，調用的是c語言的原生線程。

拿到公共數據
申請gil
python解釋器調用os原生線程
os操作cpu執行運算
當該線程執行時間到後，無論運算是否已經執行完，gil都被要求釋放
進而由其他進程重復上面的過程
等其他進程執行完後，又會切換到之前的線程（從他記錄的上下文繼續執行），整個過程是每個線程執行自己的運算，當執行時間到就進行切換（context switch）。

python下想要充分利用多核CPU，就用多進程。因為每個進程有各自獨立的GIL，互不干擾，這樣就可以真正意義上的並行執行，在python中，多進程的執行效率優於多線程(僅僅針對多核CPU而言)。

在python3.x中，GIL使用計時器（執行時間達到阈值後，當前線程釋放GIL），這樣對CPU密集型程序更加友好，但依然沒有解決GIL導致的同一時間只能執行一個線程的問題，所以效率依然不盡如人意。

對於多核CPU運行單個python進程而言，真的就是“一核有難，七核圍觀”。