Python中最常用的幾種線程鎖你會(huì)用嗎

前言本文將繼續(xù)圍繞 threading 模塊講解，基本上是純理論偏多。

對于日常開發(fā)者來講很少會(huì)使用到本文的內(nèi)容，但是對框架作者等是必備知識(shí)，同時(shí)也是高頻的面試常見問題。

線程安全線程安全是多線程或多進(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對num進(jìn)行一千萬次+1的操作

線程2對num進(jìn)行一千萬次-1的操作

結(jié)果可能會(huì)令人咋舌，num最后并不是我們所想象的結(jié)果0：

importthreading

num=0


defadd():
globalnum
foriinrange(10_000_000):
num+=1


defsub():
globalnum
foriinrange(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("numresult:%s"%num)

#結(jié)果三次采集
#numresult:669214
#numresult:-1849179
#numresult:-525674

上面這就是一個(gè)非常好的案例，想要解決這個(gè)問題就必須通過鎖來保障線程切換的時(shí)機(jī)。

需要我們值得留意的是，在Python基本數(shù)據(jù)類型中l(wèi)ist、tuple、dict本身就是屬于線程安全的，所以如果有多個(gè)線程對這3種容器做操作時(shí)，我們不必考慮線程安全問題。

鎖的作用鎖是Python提供給我們能夠自行操控線程切換的一種手段，使用鎖可以讓線程的切換變的有序。

一旦線程的切換變的有序后，各個(gè)線程之間對數(shù)據(jù)的訪問、修改就變的可控，所以若要保證線程安全，就必須使用鎖。

threading模塊中提供了5種最常見的鎖，下面是按照功能進(jìn)行劃分：

同步鎖：lock（一次只能放行一個(gè)）

遞歸鎖：rlock（一次只能放行一個(gè)）

條件鎖：condition（一次可以放行任意個(gè)）

事件鎖：event（一次全部放行）

信號量鎖：semaphore（一次可以放行特定個(gè)）

1、Lock（） 同步鎖基本介紹

Lock鎖的稱呼有很多，如：

同步鎖

互斥鎖

它們是什么意思呢？如下所示：

互斥指的是某一資源同一時(shí)刻僅能有一個(gè)訪問者對其進(jìn)行訪問，具有唯一性和排他性，但是互斥無法限制訪問者對資源的訪問順序，即訪問是無序的

同步是指在互斥的基礎(chǔ)上（大多數(shù)情況），通過其他機(jī)制實(shí)現(xiàn)訪問者對資源的有序訪問

同步其實(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)度交由其他線程。

如下所示，使用同步鎖解決最上面的問題：

importthreading

num=0


defadd():
lock.acquire()
globalnum
foriinrange(10_000_000):
num+=1
lock.release()


defsub():
lock.acquire()
globalnum
foriinrange(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("numresult:%s"%num)

#結(jié)果三次采集
#numresult:0
#numresult:0
#numresult:0

這樣這個(gè)代碼就完全變成了串行的狀態(tài)，對于這種計(jì)算密集型I/O業(yè)務(wù)來說，還不如直接使用串行化單線程執(zhí)行來得快，所以這個(gè)例子僅作為一個(gè)示例，不能概述鎖真正的用途。

死鎖現(xiàn)象

對于同步鎖來說，一次acquire()必須對應(yīng)一次release()，不能出現(xiàn)連續(xù)重復(fù)使用多次acquire()后再重復(fù)使用多次release()的操作，這樣會(huì)引起死鎖造成程序的阻塞，完全不動(dòng)了，如下所示：

importthreading

num=0


defadd():
lock.acquire()#上鎖
lock.acquire()#死鎖
#不執(zhí)行
globalnum
foriinrange(10_000_000):
num+=1
lock.release()
lock.release()


defsub():
lock.acquire()#上鎖
lock.acquire()#死鎖
#不執(zhí)行
globalnum
foriinrange(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("numresult:%s"%num)

with語句

由于threading.Lock()對象中實(shí)現(xiàn)了__enter__()與__exit__()方法，故我們可以使用with語句進(jìn)行上下文管理形式的加鎖解鎖操作：

importthreading

num=0


defadd():
withlock:
#自動(dòng)加鎖
globalnum
foriinrange(10_000_000):
num+=1
#自動(dòng)解鎖


defsub():
withlock:
#自動(dòng)加鎖
globalnum
foriinrange(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("numresult:%s"%num)

#結(jié)果三次采集
#numresult:0
#numresult:0
#numresult:0

2、RLock() 遞歸鎖

基本介紹

遞歸鎖是同步鎖的一個(gè)升級版本，在同步鎖的基礎(chǔ)上可以做到連續(xù)重復(fù)使用多次acquire()后再重復(fù)使用多次release()的操作，但是一定要注意加鎖次數(shù)和解鎖次數(shù)必須一致，否則也將引發(fā)死鎖現(xiàn)象。

下面是threading模塊與遞歸鎖提供的相關(guān)方法：

使用方式

以下是遞歸鎖的簡單使用，下面這段操作如果使用同步鎖則會(huì)發(fā)生死鎖現(xiàn)象，但是遞歸鎖不會(huì)：

importthreading

num=0


defadd():
lock.acquire()
lock.acquire()
globalnum
foriinrange(10_000_000):
num+=1
lock.release()
lock.release()


defsub():
lock.acquire()
lock.acquire()
globalnum
foriinrange(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("numresult:%s"%num)

#結(jié)果三次采集
#numresult:0
#numresult:0
#numresult:0

with語句

由于threading.RLock()對象中實(shí)現(xiàn)了__enter__()與__exit__()方法，故我們可以使用with語句進(jìn)行上下文管理形式的加鎖解鎖操作：

importthreading

num=0


defadd():
withlock:
#自動(dòng)加鎖
globalnum
foriinrange(10_000_000):
num+=1
#自動(dòng)解鎖


defsub():
withlock:
#自動(dòng)加鎖
globalnum
foriinrange(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("numresult:%s"%num)

#結(jié)果三次采集
#numresult:0
#numresult:0
#numresult: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)行：

importthreading

currentRunThreadNumber=0
maxSubThreadNumber=10


deftask():
globalcurrentRunThreadNumber
thName=threading.currentThread().name

condLock.acquire()#上鎖
print("startandwaitrunthread:%s"%thName)

condLock.wait()#暫停線程運(yùn)行、等待喚醒
currentRunThreadNumber+=1
print("carryonrunthread:%s"%thName)

condLock.release()#解鎖


if__name__=="__main__":
condLock=threading.Condition()

foriinrange(maxSubThreadNumber):
subThreadIns=threading.Thread(target=task)
subThreadIns.start()

whilecurrentRunThreadNumber"Please enter the number of threads that need to be notified to run："))

condLock.acquire()
condLock.notify(notifyNumber)#放行
condLock.release()

print("mainthreadrunend")

#先啟動(dòng)10個(gè)子線程，然后這些子線程會(huì)全部變?yōu)榈却隣顟B(tài)
#startandwaitrunthread:Thread-1
#startandwaitrunthread:Thread-2
#startandwaitrunthread:Thread-3
#startandwaitrunthread:Thread-4
#startandwaitrunthread:Thread-5
#startandwaitrunthread:Thread-6
#startandwaitrunthread:Thread-7
#startandwaitrunthread:Thread-8
#startandwaitrunthread:Thread-9
#startandwaitrunthread:Thread-10

#批量發(fā)送通知，放行特定數(shù)量的子線程繼續(xù)運(yùn)行
# Please enter the number of threads that need to be notified to run：5 #放行5個(gè)
#carryonrunthread:Thread-4
#carryonrunthread:Thread-3
#carryonrunthread:Thread-1
#carryonrunthread:Thread-2
#carryonrunthread:Thread-5

# Please enter the number of threads that need to be notified to run：5 #放行5個(gè)
#carryonrunthread:Thread-8
#carryonrunthread:Thread-10
#carryonrunthread:Thread-6
#carryonrunthread:Thread-9
#carryonrunthread:Thread-7

# Please enter the number of threads that need to be notified to run：1
#mainthreadrunend

with語句

由于threading.Condition()對象中實(shí)現(xiàn)了__enter__()與__exit__()方法，故我們可以使用with語句進(jìn)行上下文管理形式的加鎖解鎖操作：

importthreading

currentRunThreadNumber=0
maxSubThreadNumber=10


deftask():
globalcurrentRunThreadNumber
thName=threading.currentThread().name

withcondLock:
print("startandwaitrunthread:%s"%thName)
condLock.wait()#暫停線程運(yùn)行、等待喚醒
currentRunThreadNumber+=1
print("carryonrunthread:%s"%thName)


if__name__=="__main__":
condLock=threading.Condition()

foriinrange(maxSubThreadNumber):
subThreadIns=threading.Thread(target=task)
subThreadIns.start()

whilecurrentRunThreadNumber"Please enter the number of threads that need to be notified to run："))

withcondLock:
condLock.notify(notifyNumber)#放行

print("mainthreadrunend")

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ī)方式。

如下所示，我們來模擬線程和紅綠燈的操作，紅燈停，綠燈行：

importthreading

maxSubThreadNumber=3


deftask():
thName=threading.currentThread().name
print("startandwaitrunthread:%s"%thName)
eventLock.wait()#暫停運(yùn)行，等待綠燈
print("greenlight,%scarryonrun"%thName)
print("redlight,%sstoprun"%thName)
eventLock.wait()#暫停運(yùn)行，等待綠燈
print("greenlight,%scarryonrun"%thName)
print("subthread%srunend"%thName)


if__name__=="__main__":

eventLock=threading.Event()

foriinrange(maxSubThreadNumber):
subThreadIns=threading.Thread(target=task)
subThreadIns.start()

eventLock.set()#設(shè)置為綠燈
eventLock.clear()#設(shè)置為紅燈
eventLock.set()#設(shè)置為綠燈

#startandwaitrunthread:Thread-1
#startandwaitrunthread:Thread-2
#startandwaitrunthread:Thread-3

#greenlight,Thread-1carryonrun
#redlight,Thread-1stoprun
#greenlight,Thread-1carryonrun
#subthreadThread-1runend

#greenlight,Thread-3carryonrun
#redlight,Thread-3stoprun
#greenlight,Thread-3carryonrun
#subthreadThread-3runend

#greenlight,Thread-2carryonrun
#redlight,Thread-2stoprun
#greenlight,Thread-2carryonrun
#subthreadThread-2runend

5、Semaphore() 信號量鎖

基本介紹

信號量鎖也是根據(jù)條件鎖來做的，它與條件鎖和事件鎖的區(qū)別如下：

• 條件鎖：一次可以放行任意個(gè)處于“等待”狀態(tài)的線程
• 事件鎖：一次可以放行全部的處于“等待”狀態(tài)的線程
• 信號量鎖：通過規(guī)定，成批的放行特定個(gè)處于“上鎖”狀態(tài)的線程

下面是threading模塊與信號量鎖提供的相關(guān)方法：

使用方式

以下是使用示例，你可以將它當(dāng)做一段限寬的路段，每次只能放行相同數(shù)量的線程：

importthreading
importtime

maxSubThreadNumber=6


deftask():
thName=threading.currentThread().name
semaLock.acquire()
print("runsubthread%s"%thName)
time.sleep(3)
semaLock.release()


if__name__=="__main__":
#每次只能放行2個(gè)
semaLock=threading.Semaphore(2)

foriinrange(maxSubThreadNumber):
subThreadIns=threading.Thread(target=task)
subThreadIns.start()


#runsubthreadThread-1
#runsubthreadThread-2

#runsubthreadThread-3
#runsubthreadThread-4

#runsubthreadThread-6
#runsubthreadThread-5

with語句

由于threading.Semaphore()對象中實(shí)現(xiàn)了__enter__()與__exit__()方法，故我們可以使用with語句進(jìn)行上下文管理形式的加鎖解鎖操作：

importthreading
importtime

maxSubThreadNumber=6


deftask():
thName=threading.currentThread().name
withsemaLock:
print("runsubthread%s"%thName)
time.sleep(3)


if__name__=="__main__":

semaLock=threading.Semaphore(2)

foriinrange(maxSubThreadNumber):
subThreadIns=threading.Thread(target=task)
subThreadIns.start()

鎖關(guān)系淺析

上面5種鎖可以說都是基于同步鎖來做的，這些你都可以從源碼中找到答案。

首先來看RLock遞歸鎖，遞歸鎖的實(shí)現(xià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í)是有兩把鎖的，一把底層鎖（同步鎖）一把高級鎖(遞歸鎖)。

低層鎖的解鎖方式有兩種，使用wait()方法會(huì)暫時(shí)解開底層鎖同時(shí)加上一把高級鎖，只有當(dāng)接收到別的線程里的notfiy()后才會(huì)解開高級鎖和重新上鎖低層鎖，也就是說條件鎖底層是根據(jù)同步鎖和遞歸鎖的不斷切換來進(jìn)行實(shí)現(xiàn)的：

def__init__(self,lock=None):
iflockisNone:
lock=RLock()#可以看到條件鎖的內(nèi)部是基于遞歸鎖，而遞歸鎖又是基于同步鎖來做的
self._lock=lock

self.acquire=lock.acquire
self.release=lock.release
try:
self._release_save=lock._release_save
exceptAttributeError:
pass
try:
self._acquire_restore=lock._acquire_restore
exceptAttributeError:
pass
try:
self._is_owned=lock._is_owned
exceptAttributeError:
pass
self._waiters=_deque()

Event事件鎖內(nèi)部是基于條件鎖來做的：

classEvent:

def__init__(self):
self._cond=Condition(Lock())#實(shí)例化出了一個(gè)條件鎖。
self._flag=False

def_reset_internal_locks(self):
#private!calledbyThread._reset_internal_locksby_after_fork()
self._cond.__init__(Lock())

defis_set(self):
"""Returntrueifandonlyiftheinternalflagistrue."""
returnself._flag

isSet=is_set

Semaphore信號量鎖內(nèi)部也是基于條件鎖來做的：

classSemaphore:

def__init__(self,value=1):
ifvalue0:
raiseValueError("semaphoreinitialvaluemustbe>=0")
self._cond=Condition(Lock())#可以看到，這里是實(shí)例化出了一個(gè)條件鎖
self._value=value

基本練習(xí)題

條件鎖的應(yīng)用

需求：一個(gè)空列表，兩個(gè)線程輪番往里面加值（一個(gè)加偶數(shù)，一個(gè)加奇數(shù)），最終讓該列表中的值為 1 - 100 ，且是有序排列的。

importthreading

lst=[]


defeven():
"""加偶數(shù)"""
withcondLock:
foriinrange(2,101,2):
#判斷當(dāng)前列表的長度處于2是否能處盡
#如果能處盡則代表需要添加奇數(shù)
#否則就添加偶數(shù)
iflen(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()


defodd():
"""加奇數(shù)"""
withcondLock:
foriinrange(1,101,2):
iflen(lst)%2==0:
lst.append(i)
condLock.notify()
condLock.wait()
condLock.notify()


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)行對答？文本如下：

杜甫：老李啊，來喝酒！

李白：老杜啊，不喝了我喝不下了！

杜甫：老李啊，再來一壺？

杜甫：...老李？

李白：呼呼呼...睡著了..

代碼如下：

importthreading


deflibai():
event.wait()
print("李白：老杜啊，不喝了我喝不下了！")
event.set()
event.clear()
event.wait()
print("李白：呼呼呼...睡著了..")

defdufu():
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()

原文鏈接：https://www.cnblogs.com/YunyaSir/p/14963682.html