Python讀寫鎖實現實現程式碼解析
起步
Python 提供的多執行緒模型中並沒有提供讀寫鎖,讀寫鎖相對於單純的互斥鎖,適用性更高,可以多個執行緒同時佔用讀模式的讀寫鎖,但是隻能一個執行緒佔用寫模式的讀寫鎖。
通俗點說就是當沒有寫鎖時,就可以加讀鎖且任意執行緒可以同時加;而寫鎖只能有一個執行緒,且必須在沒有讀鎖時才能加上。
簡單的實現
import threading class RWlock(object): def __init__(self): self._lock = threading.Lock() self._extra = threading.Lock() self.read_num = 0 def read_acquire(self): with self._extra: self.read_num += 1 if self.read_num == 1: self._lock.acquire() def read_release(self): with self._extra: self.read_num -= 1 if self.read_num == 0: self._lock.release() def write_acquire(self): self._lock.acquire() def write_release(self): self._lock.release()
這是讀寫鎖的一個簡單的實現,self.read_num 用來儲存獲得讀鎖的執行緒數,這個屬性屬於臨界區,對其操作也要加鎖,所以這裡需要一個保護內部資料的額外的鎖 self._extra 。
但是這個鎖是不公平的。理想情況下,執行緒獲得所的機會應該是一樣的,不管執行緒是讀操作還是寫操作。而從上述程式碼可以看到,讀請求都會立即設定 self.read_num += 1,不管有沒有獲得鎖,而寫請求想要獲得鎖還得等待 read_num 為 0 。
所以這個就造成了只有鎖沒有被佔用或者沒有讀請求時,可以獲得寫許可權。我們應該想辦法避免讀模式鎖長期佔用。
讀寫鎖的優先順序
讀寫鎖也有分 讀優先 和 寫優先。上面的程式碼就屬於讀優先。
如果要改成寫優先,那就換成去記錄寫執行緒的引用計數,讀和寫在同時競爭時,可以讓寫執行緒增加寫的計數,這樣可使讀執行緒的讀鎖一直獲取不到,因為讀執行緒要先判斷寫的引用計數,若不為0,則等待其為 0,然後進行讀。這部分程式碼不羅列了。
但這樣顯然不夠靈活。我們不需要兩個相似的讀寫鎖類。我們希望重構我們程式碼,使它更強大。
改進
為了能夠滿足自定義優先順序的讀寫鎖,要記錄等待的讀寫執行緒數,並且需要兩個條件 threading.Condition 用來處理哪方優先的通知。計數引用可以擴大語義:正數:表示正在讀操作的執行緒數,負數:表示正在寫操作的執行緒數(最多-1)
在獲取讀操作時,先然後判斷時候有等待的寫執行緒,沒有,進行讀操作,有,則等待讀的計數加 1 後等待 Condition 通知;等待讀的計數減 1,計數引用加 1,繼續讀操作,若條件不成立,迴圈等待;
在獲取寫操作時,若鎖沒有被佔用,引用計數減 1,若被佔用,等待寫執行緒數加 1,等待寫條件 Condition 的通知。
讀模式和寫模式的釋放都是一樣,需要根據判斷去通知對應的 Condition:
class RWLock(object): def __init__(self): self.lock = threading.Lock() self.rcond = threading.Condition(self.lock) self.wcond = threading.Condition(self.lock) self.read_waiter = 0 # 等待獲取讀鎖的執行緒數 self.write_waiter = 0 # 等待獲取寫鎖的執行緒數 self.state = 0 # 正數:表示正在讀操作的執行緒數 負數:表示正在寫操作的執行緒數(最多-1) self.owners = [] # 正在操作的執行緒id集合 self.write_first = True # 預設寫優先,False表示讀優先 def write_acquire(self,blocking=True): # 獲取寫鎖只有當 me = threading.get_ident() with self.lock: while not self._write_acquire(me): if not blocking: return False self.write_waiter += 1 self.wcond.wait() self.write_waiter -= 1 return True def _write_acquire(self,me): # 獲取寫鎖只有當鎖沒人佔用,或者當前執行緒已經佔用 if self.state == 0 or (self.state < 0 and me in self.owners): self.state -= 1 self.owners.append(me) return True if self.state > 0 and me in self.owners: raise RuntimeError('cannot recursively wrlock a rdlocked lock') return False def read_acquire(self,blocking=True): me = threading.get_ident() with self.lock: while not self._read_acquire(me): if not blocking: return False self.read_waiter += 1 self.rcond.wait() self.read_waiter -= 1 return True def _read_acquire(self,me): if self.state < 0: # 如果鎖被寫鎖佔用 return False if not self.write_waiter: ok = True else: ok = me in self.owners if ok or not self.write_first: self.state += 1 self.owners.append(me) return True return False def unlock(self): me = threading.get_ident() with self.lock: try: self.owners.remove(me) except ValueError: raise RuntimeError('cannot release un-acquired lock') if self.state > 0: self.state -= 1 else: self.state += 1 if not self.state: if self.write_waiter and self.write_first: # 如果有寫操作在等待(預設寫優先) self.wcond.notify() elif self.read_waiter: self.rcond.notify_all() elif self.write_waiter: self.wcond.notify() read_release = unlock write_release = unlock
以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支援我們。