Python中threading庫實現線程鎖與釋放鎖

Posted on by

控制資源訪問

前文提到threading庫在多線程時，對同一資源的訪問容易導致破壞與丟失數據。為瞭保證安全的訪問一個資源對象，我們需要創建鎖。

示例如下：

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

運行之後，效果如下：

解鎖

acquire()會通過鎖進行阻塞其他線程執行中間段，release()釋放鎖，可以看到，基本都是獲得鎖之後才執行。避免瞭多個線程同時改變其資源對象，不會造成混亂。

判斷是否有另一個線程請求鎖

要確定是否有另一個線程請求鎖而不影響當前的線程，可以設置acquire()的參數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()

運行之後，效果如下：

8次

這裡，我們需要迭代很多次，work1才能獲取3次鎖。但是嘗試瞭很8次。

with lock

前文，我們通過lock.acquire()與lock.release()實現瞭鎖的獲取與釋放，但其實我們Python還給我們提供瞭一個更簡單的語法，通過with lock來獲取與釋放鎖。

示例如下：

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對象不能請求多次，即使是由同一個線程請求也不例外。如果同一個調用鏈中的多個函數訪問一個鎖，則會發生意外。如果期望在同一個線程的不同代碼需要重新獲得鎖，那麼這種情況下使用RLock。

同步線程

Condition

在實際的操作中，我們還可以使用Condition對象來同步線程。由於Condition使用瞭一個Lock，所以它可以綁定到一個共享資源，允許多個線程等待資源的更新。

示例如下：

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

運行之後，效果如下：

資源

這裡，我們通過producer線程處理完成之後調用notifyAll()，consumer等線程等到瞭它的更新，可以類比為觀察者模式。這裡是，當一個線程用完資源之後時,則會自動通知依賴它的所有線程。

屏障（barrier）

屏障是另一種線程的同步機制。barrier會建立一個控制點，所有參與的線程會在這裡阻塞，直到所有這些參與方都到達這一點。采用這種方法，線程可以單獨啟動然後暫停，直到所有線程都準備好瞭才可以繼續。

示例如下：

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

運行之後，效果如下：

屏障

從控制臺的輸出會發發現，barrier.wait()會阻塞線程，直到所有線程被創建後，才同時釋放越過這個控制點繼續執行。wait()的返回值指示瞭釋放的參與線程數，可以用來限制一些線程做清理資源等動作。

當然屏障Barrier還有一個abort()方法，該方法可以使所有等待線程接收一個BroKenBarrierError。如果線程在wait()上被阻塞而停止處理，會產生這個異常，通過except可以完成清理工作。

有限資源的並發訪問

除瞭多線程可能訪問同一個資源之外，有時候為瞭性能，我們也會限制多線程訪問同一個資源的數量。例如，線程池支持同時連接，但數據可能是固定的，或者一個網絡APP提供的並發下載數支持固定數目。這些連接就可以使用Semaphore來管理。

示例如下：

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

運行之後，效果如下：

控制臺

從圖片雖然能看所有輸出，但無法看到其停頓的事件。讀者自己運行會發現，每次頂多隻有兩個線程在工作，是因為我們設置瞭threading.Semaphore(2)。

隱藏資源

在實際的項目中，有些資源需要鎖定以便於多個線程使用，而另外一些資源則需要保護，以使它們對並非使這些資源的所有者的線程隱藏。

local()函數會創建一個對象，它能夠隱藏值，使其在不同的線程中無法被看到。示例如下：

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

運行之後，效果如下：

輸出

這裡local_data.value對所有線程都不可見，除非在某個線程中設置瞭這個屬性，這個線程才能看到它。

到此這篇關於Python中threading庫實現線程鎖與釋放鎖的文章就介紹到這瞭,更多相關Python 線程鎖與釋放鎖內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：