線程的創(chuàng)建和控制是python多線程中最重要的內(nèi)容,在python中線程控制主要使用threading庫(kù)來(lái)進(jìn)行實(shí)現(xiàn)。今天我們也以threading庫(kù)為基礎(chǔ),來(lái)介紹一下Python怎么使用threading庫(kù)實(shí)現(xiàn)線程鎖與釋放鎖。
控制資源訪問(wèn)
前文提到threading庫(kù)在多線程時(shí),對(duì)同一資源的訪問(wèn)容易導(dǎo)致破壞與丟失數(shù)據(jù)。為了保證安全的訪問(wèn)一個(gè)資源對(duì)象,我們需要?jiǎng)?chuàng)建鎖。
示例如下:
import threading import time class AddThread(): def __init__(self, start=0): self.lock = threading.Lock() self.value = start def increment(self): print("Wait Lock") self.lock.acquire() try: print("Acquire Lock") self.value += 1 print(self.value) finally: self.lock.release() def worker(a): time.sleep(1) a.increment() addThread = AddThread() for i in range(3): t = threading.Thread(target=worker, args=(addThread,)) t.start()
運(yùn)行之后,效果如下:
acquire()會(huì)通過(guò)鎖進(jìn)行阻塞其他線程執(zhí)行中間段,release()釋放鎖,可以看到,基本都是獲得鎖之后才執(zhí)行。避免了多個(gè)線程同時(shí)改變其資源對(duì)象,不會(huì)造成混亂。
判斷是否有另一個(gè)線程請(qǐng)求鎖
要確定是否有另一個(gè)線程請(qǐng)求鎖而不影響當(dāng)前的線程,可以設(shè)置acquire()的參數(shù)blocking=False。
示例如下:
import threading import time def worker2(lock): print("worker2 Wait Lock") while True: lock.acquire() try: print("Holding") time.sleep(0.5) finally: print("not Holding") lock.release() time.sleep(0.5) def worker1(lock): print("worker1 Wait Lock") num_acquire = 0 value = 0 while num_acquire < 3: time.sleep(0.5) have_it = lock.acquire(blocking=False) try: value += 1 print(value) print("Acquire Lock") if have_it: num_acquire += 1 finally: print("release Lock") if have_it: lock.release() lock = threading.Lock() word2Thread = threading.Thread( target=worker2, name='work2', args=(lock,) ) word2Thread.start() word1Thread = threading.Thread( target=worker1, name='work1', args=(lock,) ) word1Thread.start()
運(yùn)行之后,效果如下:
這里,我們需要迭代很多次,work1才能獲取3次鎖。但是嘗試了很8次。
with lock
前文,我們通過(guò)lock.acquire()與lock.release()實(shí)現(xiàn)了鎖的獲取與釋放,但其實(shí)我們Python還給我們提供了一個(gè)更簡(jiǎn)單的語(yǔ)法,通過(guò)with lock來(lái)獲取與釋放鎖。
示例如下:
import threading import time class AddThread(): def __init__(self, start=0): self.lock = threading.Lock() self.value = start def increment(self): print("Wait Lock") with self.lock: print("lock acquire") self.value += 1 print(self.value) print("lock release") def worker(a): time.sleep(1) a.increment() addThread = AddThread() for i in range(3): t = threading.Thread(target=worker, args=(addThread,)) t.start()
這里,我們只是將最上面的例子改變了一下。效果如下:
需要注意的是,正常的Lock對(duì)象不能請(qǐng)求多次,即使是由同一個(gè)線程請(qǐng)求也不例外。如果同一個(gè)調(diào)用鏈中的多個(gè)函數(shù)訪問(wèn)一個(gè)鎖,則會(huì)發(fā)生意外。如果期望在同一個(gè)線程的不同代碼需要重新獲得鎖,那么這種情況下使用RLock。
同步線程
Condition
在實(shí)際的操作中,我們還可以使用Condition對(duì)象來(lái)同步線程。由于Condition使用了一個(gè)Lock,所以它可以綁定到一個(gè)共享資源,允許多個(gè)線程等待資源的更新。
示例如下:
import threading import time def consumer(cond): print("waitCon") with cond: cond.wait() print('獲取更新的資源') def producer(cond): print("worker") with cond: print('更新資源') cond.notifyAll() cond = threading.Condition() t1 = threading.Thread(name='t1', target=consumer, args=(cond,)) t2 = threading.Thread(name='t2', target=consumer, args=(cond,)) t3 = threading.Thread(name='t3', target=producer, args=(cond,)) t1.start() time.sleep(0.2) t2.start() time.sleep(0.2) t3.start()
運(yùn)行之后,效果如下:
這里,我們通過(guò)producer線程處理完成之后調(diào)用notifyAll(),consumer等線程等到了它的更新,可以類比為觀察者模式。這里是,當(dāng)一個(gè)線程用完資源之后時(shí),則會(huì)自動(dòng)通知依賴它的所有線程。
屏障(barrier)
屏障是另一種線程的同步機(jī)制。barrier會(huì)建立一個(gè)控制點(diǎn),所有參與的線程會(huì)在這里阻塞,直到所有這些參與方都到達(dá)這一點(diǎn)。采用這種方法,線程可以單獨(dú)啟動(dòng)然后暫停,直到所有線程都準(zhǔn)備好了才可以繼續(xù)。
示例如下:
import threading import time def worker(barrier): print(threading.current_thread().getName(), "worker") worker_id = barrier.wait() print(threading.current_thread().getName(), worker_id) threads = [] barrier = threading.Barrier(3) for i in range(3): threads.append( threading.Thread( name="t" + str(i), target=worker, args=(barrier,) ) ) for t in threads: print(t.name, 'starting') t.start() time.sleep(0.1) for t in threads: t.join()
運(yùn)行之后,效果如下:
從控制臺(tái)的輸出會(huì)發(fā)發(fā)現(xiàn),barrier.wait()會(huì)阻塞線程,直到所有線程被創(chuàng)建后,才同時(shí)釋放越過(guò)這個(gè)控制點(diǎn)繼續(xù)執(zhí)行。wait()的返回值指示了釋放的參與線程數(shù),可以用來(lái)限制一些線程做清理資源等動(dòng)作。
當(dāng)然屏障Barrier還有一個(gè)abort()方法,該方法可以使所有等待線程接收一個(gè)BroKenBarrierError。如果線程在wait()上被阻塞而停止處理,會(huì)產(chǎn)生這個(gè)異常,通過(guò)except可以完成清理工作。
有限資源的并發(fā)訪問(wèn)
除了多線程可能訪問(wèn)同一個(gè)資源之外,有時(shí)候?yàn)榱诵阅?,我們也?huì)限制多線程訪問(wèn)同一個(gè)資源的數(shù)量。例如,線程池支持同時(shí)連接,但數(shù)據(jù)可能是固定的,或者一個(gè)網(wǎng)絡(luò)APP提供的并發(fā)下載數(shù)支持固定數(shù)目。這些連接就可以使用Semaphore來(lái)管理。
示例如下:
import threading import time class WorkerThread(threading.Thread): def __init__(self): super(WorkerThread, self).__init__() self.lock = threading.Lock() self.value = 0 def increment(self): with self.lock: self.value += 1 print(self.value) def worker(s, pool): with s: print(threading.current_thread().getName()) pool.increment() time.sleep(1) pool.increment() pool = WorkerThread() s = threading.Semaphore(2) for i in range(5): t = threading.Thread( name="t" + str(i), target=worker, args=(s, pool,) ) t.start()
運(yùn)行之后,效果如下:
從圖片雖然能看所有輸出,但無(wú)法看到其停頓的事件。讀者自己運(yùn)行會(huì)發(fā)現(xiàn),每次頂多只有兩個(gè)線程在工作,是因?yàn)槲覀冊(cè)O(shè)置了threading.Semaphore(2)。
隱藏資源
在實(shí)際的項(xiàng)目中,有些資源需要鎖定以便于多個(gè)線程使用,而另外一些資源則需要保護(hù),以使它們對(duì)并非使這些資源的所有者的線程隱藏。
local()函數(shù)會(huì)創(chuàng)建一個(gè)對(duì)象,它能夠隱藏值,使其在不同的線程中無(wú)法被看到。示例如下:
import threading import random def show_data(data): try: result = data.value except AttributeError: print(threading.current_thread().getName(), "No value") else: print(threading.current_thread().getName(), "value=", result) def worker(data): show_data(data) data.value = random.randint(1, 100) show_data(data) local_data = threading.local() show_data(local_data) local_data.value = 1000 show_data(local_data) for i in range(2): t = threading.Thread( name="t" + str(i), target=worker, args=(local_data,) ) t.start()
運(yùn)行之后,效果如下:
這里local_data.value對(duì)所有線程都不可見(jiàn),除非在某個(gè)線程中設(shè)置了這個(gè)屬性,這個(gè)線程才能看到它。
到此這篇Python怎么使用threading庫(kù)實(shí)現(xiàn)線程鎖與釋放鎖的文章就介紹到這里了,希望能對(duì)小伙伴有所幫助,也希望大家以后多多支持W3Cschool!