線程安全
線程安全是多線程或多進(jìn)程編程中的一個(gè)概念,在擁有共享數(shù)據(jù)的多條線程并行執(zhí)行的程序中,線程安全的代碼會(huì)通過同步機(jī)制保證各個(gè)線程都可以正常且正確的執(zhí)行,不會(huì)出現(xiàn)數(shù)據(jù)污染等意外情況。
線程安全的問題最主要還是由線程切換導(dǎo)致的,比如一個(gè)房間(進(jìn)程)中有10顆糖(資源),除此之外還有3個(gè)小人(1個(gè)主線程、2個(gè)子線程),當(dāng)小人A吃了3顆糖后被系統(tǒng)強(qiáng)制進(jìn)行休息時(shí)他認(rèn)為還剩下7顆糖,而當(dāng)小人B工作后又吃掉了3顆糖,那么當(dāng)小人A重新上崗時(shí)會(huì)認(rèn)為糖還剩下7顆,但是實(shí)際上只有4顆了。
上述例子中線程A和線程B的數(shù)據(jù)不同步,這就是線程安全問題,它可能導(dǎo)致非常嚴(yán)重的意外情況發(fā)生,我們按下面這個(gè)示例來進(jìn)行說明。
下面有一個(gè)數(shù)值num初始值為0,我們開啟2條線程:
線程1對(duì)num進(jìn)行一千萬次+1的操作
線程2對(duì)num進(jìn)行一千萬次-1的操作
結(jié)果可能會(huì)令人咋舌,num最后并不是我們所想象的結(jié)果0:
?
?
import threading num = 0 def add(): global num for i in range(10_000_000): num += 1 def sub(): global num for i in range(10_000_000): num -= 1 if __name__ == "__main__": subThread01 = threading.Thread(target=add) subThread02 = threading.Thread(target=sub) subThread01.start() subThread02.start() subThread01.join() subThread02.join() print("num result : %s" % num) # 結(jié)果三次采集 # num result : 669214 # num result : -1849179 # num result : -525674
?
?
上面這就是一個(gè)非常好的案例,想要解決這個(gè)問題就必須通過鎖來保障線程切換的時(shí)機(jī)。
需要我們值得留意的是,在Python基本數(shù)據(jù)類型中l(wèi)ist、tuple、dict本身就是屬于線程安全的,
所以如果有多個(gè)線程對(duì)這3種容器做操作時(shí),我們不必考慮線程安全問題。
鎖的作用
鎖是Python提供給我們能夠自行操控線程切換的一種手段,使用鎖可以讓線程的切換變的有序。
一旦線程的切換變的有序后,各個(gè)線程之間對(duì)數(shù)據(jù)的訪問、修改就變的可控,所以若要保證線程安全,就必須使用鎖。
threading模塊中提供了5種最常見的鎖,下面是按照功能進(jìn)行劃分:
同步鎖:lock(一次只能放行一個(gè))
遞歸鎖:rlock(一次只能放行一個(gè))
條件鎖:condition(一次可以放行任意個(gè))
事件鎖:event(一次全部放行)
信號(hào)量鎖:semaphore(一次可以放行特定個(gè))
1、Lock() 同步鎖
基本介紹
Lock鎖的稱呼有很多,如:
同步鎖
互斥鎖
它們是什么意思呢?如下所示:
互斥指的是某一資源同一時(shí)刻僅能有一個(gè)訪問者對(duì)其進(jìn)行訪問,具有唯一性和排他性,但是互斥無法限制訪問者對(duì)資源的訪問順序,即訪問是無序的
同步是指在互斥的基礎(chǔ)上(大多數(shù)情況),通過其他機(jī)制實(shí)現(xiàn)訪問者對(duì)資源的有序訪問
同步其實(shí)已經(jīng)實(shí)現(xiàn)了互斥,是互斥的一種更為復(fù)雜的實(shí)現(xiàn),因?yàn)樗诨コ獾幕A(chǔ)上實(shí)現(xiàn)了有序訪問的特點(diǎn)
下面是threading模塊與同步鎖提供的相關(guān)方法:
使用方式
同步鎖一次只能放行一個(gè)線程,一個(gè)被加鎖的線程在運(yùn)行時(shí)不會(huì)將執(zhí)行權(quán)交出去,只有當(dāng)該線程被解鎖時(shí)才會(huì)將執(zhí)行權(quán)通過系統(tǒng)調(diào)度交由其他線程。
如下所示,使用同步鎖解決最上面的問題:
?
?
import threading num = 0 def add(): lock.acquire() global num for i in range(10_000_000): num += 1 lock.release() def sub(): lock.acquire() global num for i in range(10_000_000): num -= 1 lock.release() if __name__ == "__main__": lock = threading.Lock() subThread01 = threading.Thread(target=add) subThread02 = threading.Thread(target=sub) subThread01.start() subThread02.start() subThread01.join() subThread02.join() print("num result : %s" % num) # 結(jié)果三次采集 # num result : 0 # num result : 0 # num result : 0
?
?
這樣這個(gè)代碼就完全變成了串行的狀態(tài),對(duì)于這種計(jì)算密集型I/O業(yè)務(wù)來說,還不如直接使用串行化單線程執(zhí)行來得快,所以這個(gè)例子僅作為一個(gè)示例,不能概述鎖真正的用途。
死鎖現(xiàn)象
對(duì)于同步鎖來說,一次acquire()必須對(duì)應(yīng)一次release(),不能出現(xiàn)連續(xù)重復(fù)使用多次acquire()后再重復(fù)使用多次release()的操作,這樣會(huì)引起死鎖造成程序的阻塞,完全不動(dòng)了,如下所示:
?
?
import threading num = 0 def add(): lock.acquire() # 上鎖 lock.acquire() # 死鎖 # 不執(zhí)行 global num for i in range(10_000_000): num += 1 lock.release() lock.release() def sub(): lock.acquire() # 上鎖 lock.acquire() # 死鎖 # 不執(zhí)行 global num for i in range(10_000_000): num -= 1 lock.release() lock.release() if __name__ == "__main__": lock = threading.Lock() subThread01 = threading.Thread(target=add) subThread02 = threading.Thread(target=sub) subThread01.start() subThread02.start() subThread01.join() subThread02.join() print("num result : %s" % num)
?
?
with語句
由于threading.Lock()對(duì)象中實(shí)現(xiàn)了enter__()與__exit()方法,故我們可以使用with語句進(jìn)行上下文管理形式的加鎖解鎖操作:
?
?
import threading num = 0 def add(): with lock: # 自動(dòng)加鎖 global num for i in range(10_000_000): num += 1 # 自動(dòng)解鎖 def sub(): with lock: # 自動(dòng)加鎖 global num for i in range(10_000_000): num -= 1 # 自動(dòng)解鎖 if __name__ == "__main__": lock = threading.Lock() subThread01 = threading.Thread(target=add) subThread02 = threading.Thread(target=sub) subThread01.start() subThread02.start() subThread01.join() subThread02.join() print("num result : %s" % num) # 結(jié)果三次采集 # num result : 0 # num result : 0 # num result : 0
?
?
2、RLock() 遞歸鎖
基本介紹
遞歸鎖是同步鎖的一個(gè)升級(jí)版本,在同步鎖的基礎(chǔ)上可以做到連續(xù)重復(fù)使用多次acquire()后再重復(fù)使用多次release()的操作,但是一定要注意加鎖次數(shù)和解鎖次數(shù)必須一致,否則也將引發(fā)死鎖現(xiàn)象。
下面是threading模塊與遞歸鎖提供的相關(guān)方法:
使用方式
以下是遞歸鎖的簡(jiǎn)單使用,下面這段操作如果使用同步鎖則會(huì)發(fā)生死鎖現(xiàn)象,但是遞歸鎖不會(huì):
?
?
import threading num = 0 def add(): lock.acquire() lock.acquire() global num for i in range(10_000_000): num += 1 lock.release() lock.release() def sub(): lock.acquire() lock.acquire() global num for i in range(10_000_000): num -= 1 lock.release() lock.release() if __name__ == "__main__": lock = threading.RLock() subThread01 = threading.Thread(target=add) subThread02 = threading.Thread(target=sub) subThread01.start() subThread02.start() subThread01.join() subThread02.join() print("num result : %s" % num) # 結(jié)果三次采集 # num result : 0 # num result : 0 # num result : 0
?
?
with語句
由于threading.RLock()對(duì)象中實(shí)現(xiàn)了enter__()與__exit()方法,故我們可以使用with語句進(jìn)行上下文管理形式的加鎖解鎖操作:
?
?
import threading num = 0 def add(): with lock: # 自動(dòng)加鎖 global num for i in range(10_000_000): num += 1 # 自動(dòng)解鎖 def sub(): with lock: # 自動(dòng)加鎖 global num for i in range(10_000_000): num -= 1 # 自動(dòng)解鎖 if __name__ == "__main__": lock = threading.RLock() subThread01 = threading.Thread(target=add) subThread02 = threading.Thread(target=sub) subThread01.start() subThread02.start() subThread01.join() subThread02.join() print("num result : %s" % num) # 結(jié)果三次采集 # num result : 0 # num result : 0 # num result : 0
?
?
3、Condition() 條件鎖
基本介紹
條件鎖是在遞歸鎖的基礎(chǔ)上增加了能夠暫停線程運(yùn)行的功能。并且我們可以使用wait()與notify()來控制線程執(zhí)行的個(gè)數(shù)。
注意:條件鎖可以自由設(shè)定一次放行幾個(gè)線程。
下面是threading模塊與條件鎖提供的相關(guān)方法:
使用方式
下面這個(gè)案例會(huì)啟動(dòng)10個(gè)子線程,并且會(huì)立即將10個(gè)子線程設(shè)置為等待狀態(tài)。
然后我們可以發(fā)送一個(gè)或者多個(gè)通知,來恢復(fù)被等待的子線程繼續(xù)運(yùn)行:
?
?
import threading currentRunThreadNumber = 0 maxSubThreadNumber = 10 def task(): global currentRunThreadNumber thName = threading.currentThread().name condLock.acquire() # 上鎖 print("start and wait run thread : %s" % thName) condLock.wait() # 暫停線程運(yùn)行、等待喚醒 currentRunThreadNumber += 1 print("carry on run thread : %s" % thName) condLock.release() # 解鎖 if __name__ == "__main__": condLock = threading.Condition() for i in range(maxSubThreadNumber): subThreadIns = threading.Thread(target=task) subThreadIns.start() while currentRunThreadNumber < maxSubThreadNumber: notifyNumber = int( input("Please enter the number of threads that need to be notified to run:")) condLock.acquire() condLock.notify(notifyNumber) # 放行 condLock.release() print("main thread run end") # 先啟動(dòng)10個(gè)子線程,然后這些子線程會(huì)全部變?yōu)榈却隣顟B(tài) # start and wait run thread : Thread-1 # start and wait run thread : Thread-2 # start and wait run thread : Thread-3 # start and wait run thread : Thread-4 # start and wait run thread : Thread-5 # start and wait run thread : Thread-6 # start and wait run thread : Thread-7 # start and wait run thread : Thread-8 # start and wait run thread : Thread-9 # start and wait run thread : Thread-10 # 批量發(fā)送通知,放行特定數(shù)量的子線程繼續(xù)運(yùn)行 # Please enter the number of threads that need to be notified to run:5 # 放行5個(gè) # carry on run thread : Thread-4 # carry on run thread : Thread-3 # carry on run thread : Thread-1 # carry on run thread : Thread-2 # carry on run thread : Thread-5 #學(xué)習(xí)中遇到問題沒人解答?小編創(chuàng)建了一個(gè)Python學(xué)習(xí)交流群:725638078 # Please enter the number of threads that need to be notified to run:5 # 放行5個(gè) # carry on run thread : Thread-8 # carry on run thread : Thread-10 # carry on run thread : Thread-6 # carry on run thread : Thread-9 # carry on run thread : Thread-7 # Please enter the number of threads that need to be notified to run:1 # main thread run end
?
?
with語句
由于threading.Condition()對(duì)象中實(shí)現(xiàn)了enter__()與__exit()方法,故我們可以使用with語句進(jìn)行上下文管理形式的加鎖解鎖操作:
?
?
import threading currentRunThreadNumber = 0 maxSubThreadNumber = 10 def task(): global currentRunThreadNumber thName = threading.currentThread().name with condLock: print("start and wait run thread : %s" % thName) condLock.wait() # 暫停線程運(yùn)行、等待喚醒 currentRunThreadNumber += 1 print("carry on run thread : %s" % thName) if __name__ == "__main__": condLock = threading.Condition() for i in range(maxSubThreadNumber): subThreadIns = threading.Thread(target=task) subThreadIns.start() while currentRunThreadNumber < maxSubThreadNumber: notifyNumber = int( input("Please enter the number of threads that need to be notified to run:")) with condLock: condLock.notify(notifyNumber) # 放行 print("main thread run end")
?
?
4、Event() 事件鎖
基本介紹
事件鎖是基于條件鎖來做的,它與條件鎖的區(qū)別在于一次只能放行全部,不能放行任意個(gè)數(shù)量的子線程繼續(xù)運(yùn)行。
我們可以將事件鎖看為紅綠燈,當(dāng)紅燈時(shí)所有子線程都暫停運(yùn)行,并進(jìn)入“等待”狀態(tài),當(dāng)綠燈時(shí)所有子線程都恢復(fù)“運(yùn)行”。
下面是threading模塊與事件鎖提供的相關(guān)方法:
使用方式
事件鎖不能利用with語句來進(jìn)行使用,只能按照常規(guī)方式。
如下所示,我們來模擬線程和紅綠燈的操作,紅燈停,綠燈行:
?
?
# 生成一個(gè)事件鎖對(duì)象 eve = threading.Event() # 將事件鎖設(shè)置為紅燈狀態(tài) eve.clear() # 判斷事件鎖的狀態(tài) eve.is_set() # 將當(dāng)前線程設(shè)置’等待‘狀態(tài) eve.wait() # 將事件鎖設(shè)置為綠燈狀態(tài) eve.set() import time import threading def light(eve): print(f'當(dāng)前時(shí)間:{time.ctime()}, 紅燈還有 5s 結(jié)束!') time.sleep(5) print(f'當(dāng)前時(shí)間:{time.ctime()}, 綠燈亮!') eve.set() # 設(shè)置事件鎖標(biāo)志為 True def car(eve, name): print(f'當(dāng)前時(shí)間:{time.ctime()}, 車 {name} 正在等紅燈') eve.wait() # 將當(dāng)前線程設(shè)置為等待狀態(tài),等待事件鎖標(biāo)志為 True 再執(zhí)行 print(f'當(dāng)前時(shí)間:{time.ctime()}, 車 {name} 開始通行') if __name__ == '__main__': eve = threading.Event() # 事件鎖默認(rèn)標(biāo)志為 False t1 = threading.Thread(target=light, args=(eve,)) t1.start() for each in 'ABCDE': t2 = threading.Thread(target=car, args=(eve, each)) t2.start() #學(xué)習(xí)中遇到問題沒人解答?小編創(chuàng)建了一個(gè)Python學(xué)習(xí)交流群:725638078 # 執(zhí)行看一下打印結(jié)果 當(dāng)前時(shí)間:Fri Jul 29 11:32:58 2022, 紅燈還有 5s 結(jié)束! 當(dāng)前時(shí)間:Fri Jul 29 11:32:58 2022, 車 A 正在等紅燈 當(dāng)前時(shí)間:Fri Jul 29 11:32:58 2022, 車 B 正在等紅燈 當(dāng)前時(shí)間:Fri Jul 29 11:32:58 2022, 車 C 正在等紅燈 當(dāng)前時(shí)間:Fri Jul 29 11:32:58 2022, 車 D 正在等紅燈 當(dāng)前時(shí)間:Fri Jul 29 11:32:58 2022, 車 E 正在等紅燈 當(dāng)前時(shí)間:Fri Jul 29 11:33:03 2022, 綠燈亮! 當(dāng)前時(shí)間:Fri Jul 29 11:33:03 2022, 車 C 開始通行 當(dāng)前時(shí)間:Fri Jul 29 11:33:03 2022, 車 B 開始通行 當(dāng)前時(shí)間:Fri Jul 29 11:33:03 2022, 車 E 開始通行 當(dāng)前時(shí)間:Fri Jul 29 11:33:03 2022, 車 A 開始通行 當(dāng)前時(shí)間:Fri Jul 29 11:33:03 2022, 車 D 開始通行
?
?
5、Semaphore() 信號(hào)量鎖
基本介紹
信號(hào)量鎖也是根據(jù)條件鎖來做的,它與條件鎖和事件鎖的區(qū)別如下:
條件鎖:一次可以放行任意個(gè)處于“等待”狀態(tài)的線程
事件鎖:一次可以放行全部的處于“等待”狀態(tài)的線程
信號(hào)量鎖:通過規(guī)定,成批的放行特定個(gè)處于“上鎖”狀態(tài)的線程
下面是threading模塊與信號(hào)量鎖提供的相關(guān)方法:
使用方式
以下是使用示例,你可以將它當(dāng)做一段限寬的路段,每次只能放行相同數(shù)量的線程:
?
?
import threading import time maxSubThreadNumber = 6 def task(): thName = threading.currentThread().name semaLock.acquire() print("run sub thread %s" % thName) time.sleep(3) semaLock.release() if __name__ == "__main__": # 每次只能放行2個(gè) semaLock = threading.Semaphore(2) for i in range(maxSubThreadNumber): subThreadIns = threading.Thread(target=task) subThreadIns.start() # run sub thread Thread-1 # run sub thread Thread-2 # run sub thread Thread-3 # run sub thread Thread-4 # run sub thread Thread-6 # run sub thread Thread-5
?
?
with語句
由于threading.Semaphore()對(duì)象中實(shí)現(xiàn)了enter__()與__exit()方法,故我們可以使用with語句進(jìn)行上下文管理形式的加鎖解鎖操作:
?
?
import threading import time maxSubThreadNumber = 6 def task(): thName = threading.currentThread().name with semaLock: print("run sub thread %s" % thName) time.sleep(3) if __name__ == "__main__": semaLock = threading.Semaphore(2) for i in range(maxSubThreadNumber): subThreadIns = threading.Thread(target=task) subThreadIns.start()
?
?
鎖關(guān)系淺析
上面5種鎖可以說都是基于同步鎖來做的,這些你都可以從源碼中找到答案。
首先來看RLock遞歸鎖,遞歸鎖的實(shí)現(xiàn)非常簡(jiǎn)單,它的內(nèi)部會(huì)維護(hù)著一個(gè)計(jì)數(shù)器,當(dāng)計(jì)數(shù)器不為0的時(shí)候該線程不能被I/O操作和時(shí)間輪詢機(jī)制切換。但是當(dāng)計(jì)數(shù)器為0的時(shí)候便不會(huì)如此了:
?
?
def __init__(self): self._block = _allocate_lock() self._owner = None self._count = 0 # 計(jì)數(shù)器
?
?
而Condition條件鎖的內(nèi)部其實(shí)是有兩把鎖的,一把底層鎖(同步鎖)一把高級(jí)鎖(遞歸鎖)。
低層鎖的解鎖方式有兩種,使用wait()方法會(huì)暫時(shí)解開底層鎖同時(shí)加上一把高級(jí)鎖,只有當(dāng)接收到別的線程里的notfiy()后才會(huì)解開高級(jí)鎖和重新上鎖低層鎖,也就是說條件鎖底層是根據(jù)同步鎖和遞歸鎖的不斷切換來進(jìn)行實(shí)現(xiàn)的:
?
?
def __init__(self, lock=None): if lock is None: lock = RLock() # 可以看到條件鎖的內(nèi)部是基于遞歸鎖,而遞歸鎖又是基于同步鎖來做的 self._lock = lock self.acquire = lock.acquire self.release = lock.release try: self._release_save = lock._release_save except AttributeError: pass try: self._acquire_restore = lock._acquire_restore except AttributeError: pass try: self._is_owned = lock._is_owned except AttributeError: pass self._waiters = _deque()
?
?
基本練習(xí)題
條件鎖的應(yīng)用
需求:一個(gè)空列表,兩個(gè)線程輪番往里面加值(一個(gè)加偶數(shù),一個(gè)加奇數(shù)),最終讓該列表中的值為 1 - 100 ,且是有序排列的。
?
?
import threading lst = [] def even(): """加偶數(shù)""" with condLock: for i in range(2, 101, 2): # 判斷當(dāng)前列表的長(zhǎng)度處于2是否能處盡 # 如果能處盡則代表需要添加奇數(shù) # 否則就添加偶數(shù) if len(lst) % 2 != 0: # 添偶數(shù) lst.append(i) # 先添加值 condLock.notify() # 告訴另一個(gè)線程,你可以加奇數(shù)了,但是這里不會(huì)立即交出執(zhí)行權(quán) condLock.wait() # 交出執(zhí)行權(quán),并等待另一個(gè)線程通知加偶數(shù) else: # 添奇數(shù) condLock.wait() # 交出執(zhí)行權(quán),等待另一個(gè)線程通知加偶數(shù) lst.append(i) condLock.notify() condLock.notify() def odd(): """加奇數(shù)""" with condLock: for i in range(1, 101, 2): if len(lst) % 2 == 0: lst.append(i) condLock.notify() condLock.wait() condLock.notify() #學(xué)習(xí)中遇到問題沒人解答?小編創(chuàng)建了一個(gè)Python學(xué)習(xí)交流群:725638078 if __name__ == "__main__": condLock = threading.Condition() addEvenTask = threading.Thread(target=even) addOddTask = threading.Thread(target=odd) addEvenTask.start() addOddTask.start() addEvenTask.join() addOddTask.join() print(lst)
?
?
事件鎖的應(yīng)用
有2個(gè)任務(wù)線程來扮演李白和杜甫,如何讓他們一人一句進(jìn)行對(duì)答?文本如下:
杜甫:老李啊,來喝酒!
李白:老杜啊,不喝了我喝不下了!
杜甫:老李啊,再來一壺?
杜甫:…老李?
李白:呼呼呼…睡著了..
代碼如下:
?
?
import threading def libai(): event.wait() print("李白:老杜啊,不喝了我喝不下了!") event.set() event.clear() event.wait() print("李白:呼呼呼...睡著了..") def dufu(): print("杜甫:老李啊,來喝酒!") event.set() event.clear() event.wait() print("杜甫:老李啊,再來一壺?") print("杜甫:...老李?") event.set() if __name__ == '__main__': event = threading.Event() t1 = threading.Thread(target=libai) t2 = threading.Thread(target=dufu) t1.start() t2.start() t1.join() t2.join()
?
?
審核編輯:黃飛
?
評(píng)論