互斥鎖、死鎖現象、遞歸鎖、信號量

一、互斥鎖

互斥鎖也叫用戶鎖、同步鎖。 ----------同一時間只有一個任務被執行
在多進程/多線程程序中，當多個線程處理一個公共數據時，會有數據安全問題

唯一能保證數據安全的，就是通過加鎖的方式，同一時間只能有一個修改數據的操作，將處理數據變為串行。雖然犧牲了速度，但是保證了數據安全。

來看一個不加鎖的例子：

from threading import Thread
import time

def task():
    global num
    temp = num
    time.sleep(0.001)    
    num = temp -1

num = 100

t_l = []
for i in range(100):
    t = threading.Thread(target=task)
    t.start()
    t_l.append(t)

for t in t_l:
    t.join()

print(num)

在上面這個程序中，我們開一百個線程，每個線程都對全局變量num實現-1的操作，如果順利，最終num的值應該為0.
實際運行過程是這樣的：
100個線程開始搶GIL，搶到的將被CPU執行：
step1: 執行global num
step2: temp = num 賦值操作
step3: 發生I/O阻塞，掛起，GIL釋放 （下一步的num=temp-1 還未被執行，因此全局變量num的值仍然為100）
剩余的99個線程搶GIL鎖，重復上面的步驟。
剩余的98個線程搶GIL鎖，重復上面的步驟。
。。。
如果阻塞時間夠長（比如大於0.1秒），在阻塞期間，100個線程都被切換一遍的話，那么最終num的值是99；
如果阻塞時間短一點，在某個時刻，前面阻塞的線程恢復並搶到了GIL被CPU繼續執行，那么執行num=temp-1賦值操作 ，全局變量num的值被改變，線程結束，下一個被執行的線程拿到的num值就是99……依次類推，最終num的值經過多次賦值操作后將變得不確定，這取決於有多少線程從阻塞中恢復過來。
如果不阻塞的話，每個線程都會執行對num 賦值操作，下一個線程拿到的num就是上一個線程減一的結果，最終num的值歸零。

下面我們進行加鎖操作：

lock = Lock() # 獲取鎖對象 
lock.acquire() # 加鎖 
數據操作部分 
lock.release() # 釋放鎖

from threading import Thread
import time

def task(lock):
    global num

    lock.acquire()  #獲得鎖

    temp = num
    time.sleep(0.01)   #加鎖之后，無論中間睡多長時間，最終修改的結果0，因為此時加鎖部分變成嚴格意義的串行
    num = temp -1

    lock.release()  # 執行完數據修改，釋放鎖

num = 100

lock = Lock() 
# 實例化一個用戶鎖/互斥鎖，這個鎖是全局變量，
# 每個線程獲取到鎖才能執行，執行完了釋放，下一個線程才能獲取鎖
t_l = []
for i in range(100):
    t = Thread(target=task,args=(lock,))
    t.start()
    t_l.append(t)

for t in t_l:
    t.join()

print(num)

上面加鎖和解鎖的操作也可以通過上下文管理來實現：

with lock:
    數據修改部分

 

二、死鎖現象和遞歸鎖

兩個或兩個以上的進程或線程在執行過程中，因爭奪資源而造成的一種互相等待的現象，若無外力作用，它們都將無法推進下去。此時稱系統處於死鎖狀態或系統產生了死鎖，這些永遠在互相等待的進程稱為死鎖進程。看一個栗子：

from threading import Thread,Lock
import time
LockA=Lock()
LockB=Lock()

class MyThread(Thread):

    def run(self):
        self.foo()
        self.bar()

    def foo(self):
        LockA.acquire()
        print('%s got LockA '% self.name)
        LockB.acquire()
        print('%s got LockB'%self.name)
        LockB.release()
        LockA.release()

        # Thread-1在執行完foo函數后，釋放鎖。然后繼續執行bar函數，重新獲取鎖。

    def bar(self):
        LockB.acquire()
        print('%s got LockB ' % self.name)
        time.sleep(1)   
        # 讓Thread-1獲得了LockB后阻塞，OS切換線程Thread-2，其先執行foo,獲取到LockA，然后需要獲取LockB，才能執行下去，才能釋放鎖。而LockB在Thread-1手中，Thread-1從阻塞中恢復，需要獲得LockA才能繼續執行下去，才能釋放鎖。於是兩個線程互相等待，發生死鎖。
        LockA.acquire() # 因為LockA已經被其它線程搶走了，所以這里卡死了。
        print('%s got LockA' % self.name)
        LockA.release()
        LockB.release()
for i in range(10):
    t = MyThread()
    t.start()

'''死鎖了
Thread-1 got LockA 
Thread-1 got LockB
Thread-1 got LockB  
Thread-2 got LockA 
'''

解決方案就是使用遞歸鎖：
在Python中為了支持在同一線程中多次請求同一資源，python提供了可重入鎖RLock。
這個RLock內部維護着一個Lock和一個counter變量，counter記錄了acquire的次數，從而使得資源可以被多次require。直到一個線程所有的acquire都被release，其他的線程才能獲得資源。上面的例子如果使用RLock代替Lock，則不會發生死鎖：
rlock = threading.RLock() # 拿到一個可重入鎖對象，將上面的所有鎖都更換為rlock。

三、信號量

互斥鎖同時只允許一個線程更改數據，而Semaphore是同時允許一定數量的線程更改數據。
Semaphore管理一個內置的計數器，
每當調用acquire()時內置計數器-1；
調用release() 時內置計數器+1；
計數器不能小於0；當計數器為0時，acquire()將阻塞線程直到其他線程調用release()。
鎖信號量與進程池的概念很像，但是要區分開，信號量涉及到加鎖的概念。
看一個栗子：

import threading import Thread,Semaphore
import multiprocessing import Process,Semaphore
import time, os, random

def go_wc(sem,name):
    with sem:
        print('員工%s 搶到一個茅坑,開始蹲...'% name)
        time.sleep(random.uniform(1,3))
        print('員工%s 完事了,感到身心愉悅...'% name)

if __name__ == '__main__':  # 如果是進程的話，在windows系統下，進程必須寫到if __name__ == '__main__':內，否則報錯
    print('大家開始上廁所》》》')
    sem = Semaphore(3)    # 設定最大為3
    # sem = Semaphore(3)

    t_l = []
    for i in range(10):
        t = Thread(target=go_wc, args=(sem,i))
        # t = Process(target=go_wc,args=(sem,i))
        t.start()
        t_l.append(t)

    for t in t_l:
        t.join()

    print('大家都完事了《《《')

'''
大家開始上廁所》》》
員工1 搶到一個茅坑,開始蹲...
員工0 搶到一個茅坑,開始蹲...
員工5 搶到一個茅坑,開始蹲...
員工0 完事了,感到身心愉悅...   同一時刻，只有3個坑，其中一個完事了，下一個才能開始
員工4 搶到一個茅坑,開始蹲...
'''

 

 

四、Event事件

在多線程環境中，每個線程的執行一般是獨立的，如果一個線程的執行依賴於另一個線程的狀態，那么就有必要引入某種標志位來進行判斷，event就相當於一個全局的標志位。event常用於主線程控制其他線程的執行。
創建一個event對象：
event = threading.Event()
event對象的方法：
1. event.isSet() 或 event.is_set(), 返回event對象的bool值，event對象的初始bool值是False.
2. event.wait() 如果上面是True, 啥也不做，往下執行，如果上面是False, 則阻塞線程. wait(num)為超時設置，超過num秒，繼續往下執行。
3. event.set() 設置event對象True
4. event.clear() 恢復為False
5. 圖示：

舉例說明：

import threading
import time

def request():
    print('waitting for server...')
    event.wait()  #阻塞，等待主線程開啟服務器
    print('connecting to server....')

if __name__ == '__main__':
    event = threading.Event()
    for i in range(5):
        t = threading.Thread(target=request)
        t.start()

    print('attemp to start server')
    time.sleep(3)
    event.set() # 開啟服務器后，更改event狀態