詳解 5 種 Python 線程鎖

大家好，今天本文將圍繞 threading 模塊講解，基本上是純理論偏多。

對於日常開發者來講，這些內容是必備，同時也是高頻的面試常見問題。

官方文檔 (https://docs.python.org/zh-cn/3.6/library/threading.html)

線程安全

線程安全是多線程或多進程編程中的一個概念，在擁有共享數據的多條線程並行執行的程序中，線程安全的代碼會通過同步機制保證各個線程都可以正常且正確的執行，不會出現數據污染等意外情況。

線程安全的問題最主要還是由線程切換導致的，比如一個房間（進程）中有 10 顆糖（資源），除此之外還有 3 個小人（1 個主線程、2 個子線程），當小人 A 吃了 3 顆糖後被系統強制進行休息時他認爲還剩下 7 顆糖，而當小人 B 工作後又喫掉了 3 顆糖，那麼當小人 A 重新上崗時會認爲糖還剩下 7 顆，但是實際上只有 4 顆了。

上述例子中線程 A 和線程 B 的數據不同步，這就是線程安全問題，它可能導致非常嚴重的意外情況發生，我們按下面這個示例來進行說明。

下面有一個數值 num 初始值爲 0，我們開啓 2 條線程：

線程 1 對 num 進行一千萬次 + 1 的操作
線程 2 對 num 進行一千萬次 - 1 的操作

結果可能會令人咋舌，num 最後並不是我們所想象的結果 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)

# 結果三次採集
# num result : 669214
# num result : -1849179
# num result : -525674

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

需要我們值得留意的是，在 Python 基本數據類型中 list、tuple、dict 本身就是屬於線程安全的，所以如果有多個線程對這 3 種容器做操作時，我們不必考慮線程安全問題。

鎖的作用

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

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

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

同步鎖：lock（一次只能放行一個）
遞歸鎖：rlock（一次只能放行一個）
條件鎖：condition（一次可以放行任意個）
事件鎖：event（一次全部放行）
信號量鎖：semaphore（一次可以放行特定個）

1、Lock() 同步鎖

基本介紹

Lock 鎖的稱呼有很多，如：

同步鎖
互斥鎖

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

互斥指的是某一資源同一時刻僅能有一個訪問者對其進行訪問，具有唯一性和排他性，但是互斥無法限制訪問者對資源的訪問順序，即訪問是無序的
同步是指在互斥的基礎上（大多數情況），通過其他機制實現訪問者對資源的有序訪問
同步其實已經實現了互斥，是互斥的一種更爲複雜的實現，因爲它在互斥的基礎上實現了有序訪問的特點

下面是 threading 模塊與同步鎖提供的相關方法：

tIDYkL

使用方式

同步鎖一次只能放行一個線程，一個被加鎖的線程在運行時不會將執行權交出去，只有當該線程被解鎖時纔會將執行權通過系統調度交由其他線程。

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

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)

# 結果三次採集
# num result : 0
# num result : 0
# num result : 0

這樣這個代碼就完全變成了串行的狀態，對於這種計算密集型 I/O 業務來說，還不如直接使用串行化單線程執行來得快，所以這個例子僅作爲一個示例，不能概述鎖真正的用途。

死鎖現象

對於同步鎖來說，一次 acquire() 必須對應一次 release()，不能出現連續重複使用多次 acquire() 後再重複使用多次 release() 的操作，這樣會引起死鎖造成程序的阻塞，完全不動了，如下所示：

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.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() 對象中實現了__enter__() 與__exit__() 方法，故我們可以使用 with 語句進行上下文管理形式的加鎖解鎖操作：

import threading

num = 0


def add():
    with lock:
        # 自動加鎖
        global num
        for i in range(10_000_000):
            num += 1
        # 自動解鎖


def sub():
    with lock:
        # 自動加鎖
        global num
        for i in range(10_000_000):
            num -= 1
        # 自動解鎖


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)
    
# 結果三次採集
# num result : 0
# num result : 0
# num result : 0

2、RLock() 遞歸鎖

基本介紹

遞歸鎖是同步鎖的一個升級版本，在同步鎖的基礎上可以做到連續重複使用多次 acquire() 後再重複使用多次 release() 的操作，但是一定要注意加鎖次數和解鎖次數必須一致，否則也將引發死鎖現象。

下面是 threading 模塊與遞歸鎖提供的相關方法：

HoBqdI

使用方式

以下是遞歸鎖的簡單使用，下面這段操作如果使用同步鎖則會發生死鎖現象，但是遞歸鎖不會：

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)

# 結果三次採集
# num result : 0
# num result : 0
# num result : 0

with 語句

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

import threading

num = 0


def add():
    with lock:
        # 自動加鎖
        global num
        for i in range(10_000_000):
            num += 1
        # 自動解鎖


def sub():
    with lock:
        # 自動加鎖
        global num
        for i in range(10_000_000):
            num -= 1
        # 自動解鎖


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)

# 結果三次採集
# num result : 0
# num result : 0
# num result : 0

3、Condition() 條件鎖

基本介紹

條件鎖是在遞歸鎖的基礎上增加了能夠暫停線程運行的功能。並且我們可以使用 wait() 與 notify() 來控制線程執行的個數。

注意：條件鎖可以自由設定一次放行幾個線程。

下面是 threading 模塊與條件鎖提供的相關方法：

ruyJ20

使用方式

下面這個案例會啓動 10 個子線程，並且會立即將 10 個子線程設置爲等待狀態。

然後我們可以發送一個或者多個通知，來恢復被等待的子線程繼續運行：

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()  # 暫停線程運行、等待喚醒
    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")
    
# 先啓動10個子線程，然後這些子線程會全部變爲等待狀態
# 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

# 批量發送通知，放行特定數量的子線程繼續運行
# Please enter the number of threads that need to be notified to run：5  # 放行5個
# 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

# Please enter the number of threads that need to be notified to run：5  # 放行5個
# 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() 對象中實現了__enter__() 與__exit__() 方法，故我們可以使用 with 語句進行上下文管理形式的加鎖解鎖操作：

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()  # 暫停線程運行、等待喚醒
        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() 事件鎖

基本介紹

事件鎖是基於條件鎖來做的，它與條件鎖的區別在於一次只能放行全部，不能放行任意個數量的子線程繼續運行。

我們可以將事件鎖看爲紅綠燈，當紅燈時所有子線程都暫停運行，並進入 “等待” 狀態，當綠燈時所有子線程都恢復“運行”。

下面是 threading 模塊與事件鎖提供的相關方法：

DV1pko

使用方式

事件鎖不能利用 with 語句來進行使用，只能按照常規方式。

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

import threading

maxSubThreadNumber = 3


def task():
    thName = threading.currentThread().name
    print("start and wait run thread : %s" % thName)
    eventLock.wait()  # 暫停運行，等待綠燈
    print("green light, %s carry on run" % thName)
    print("red light, %s stop run" % thName)
    eventLock.wait()  # 暫停運行，等待綠燈
    print("green light, %s carry on run" % thName)
    print("sub thread %s run end" % thName)


if __name__ == "__main__":

    eventLock = threading.Event()

    for i in range(maxSubThreadNumber):
        subThreadIns = threading.Thread(target=task)
        subThreadIns.start()

    eventLock.set()  # 設置爲綠燈
    eventLock.clear()  # 設置爲紅燈
    eventLock.set()  # 設置爲綠燈

# start and wait run thread : Thread-1
# start and wait run thread : Thread-2
# start and wait run thread : Thread-3

# green light, Thread-1 carry on run
# red light, Thread-1 stop run
# green light, Thread-1 carry on run
# sub thread Thread-1 run end

# green light, Thread-3 carry on run
# red light, Thread-3 stop run
# green light, Thread-3 carry on run
# sub thread Thread-3 run end

# green light, Thread-2 carry on run
# red light, Thread-2 stop run
# green light, Thread-2 carry on run
# sub thread Thread-2 run end

5、Semaphore() 信號量鎖

基本介紹

信號量鎖也是根據條件鎖來做的，它與條件鎖和事件鎖的區別如下：

條件鎖：一次可以放行任意個處於 “等待” 狀態的線程
事件鎖：一次可以放行全部的處於 “等待” 狀態的線程
信號量鎖：通過規定，成批的放行特定個處於 “上鎖” 狀態的線程

下面是 threading 模塊與信號量鎖提供的相關方法：

vGv6eO

使用方式

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

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個
    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() 對象中實現了__enter__() 與__exit__() 方法，故我們可以使用 with 語句進行上下文管理形式的加鎖解鎖操作：

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

鎖關係淺析

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

首先來看 RLock 遞歸鎖，遞歸鎖的實現非常簡單，它的內部會維護着一個計數器，當計數器不爲 0 的時候該線程不能被 I/O 操作和時間輪詢機制切換。但是當計數器爲 0 的時候便不會如此了：

def __init__(self):
    self._block = _allocate_lock()
    self._owner = None
    self._count = 0  # 計數器

而 Condition 條件鎖的內部其實是有兩把鎖的，一把底層鎖（同步鎖）一把高級鎖 (遞歸鎖)。

低層鎖的解鎖方式有兩種，使用 wait() 方法會暫時解開底層鎖同時加上一把高級鎖，只有當接收到別的線程裏的 notfiy() 後纔會解開高級鎖和重新上鎖低層鎖，也就是說條件鎖底層是根據同步鎖和遞歸鎖的不斷切換來進行實現的：

def __init__(self, lock=None):
    if lock is None:
        lock = RLock()  # 可以看到條件鎖的內部是基於遞歸鎖，而遞歸鎖又是基於同步鎖來做的
    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()

Event 事件鎖內部是基於條件鎖來做的：

class Event:

    def __init__(self):
        self._cond = Condition(Lock())  # 實例化出了一個條件鎖。
        self._flag = False

    def _reset_internal_locks(self):
        # private!  called by Thread._reset_internal_locks by _after_fork()
        self._cond.__init__(Lock())

    def is_set(self):
        """Return true if and only if the internal flag is true."""
        return self._flag

    isSet = is_set

Semaphore 信號量鎖內部也是基於條件鎖來做的：

class Semaphore:

    def __init__(self, value=1):
        if value < 0:
            raise ValueError("semaphore initial value must be >= 0")
        self._cond = Condition(Lock()) # 可以看到，這裏是實例化出了一個條件鎖
        self._value = value

基本練習題

條件鎖的應用

需求：一個空列表，兩個線程輪番往裏面加值（一個加偶數，一個加奇數），最終讓該列表中的值爲 1 - 100 ，且是有序排列的。

import threading

lst = []


def even():
    """加偶數"""
    with condLock:
        for i in range(2, 101, 2):
            # 判斷當前列表的長度處於2是否能處盡
            # 如果能處盡則代表需要添加奇數
            # 否則就添加偶數
            if len(lst) % 2 != 0:
                # 添偶數
                lst.append(i)      # 先添加值
                condLock.notify()  # 告訴另一個線程，你可以加奇數了，但是這裏不會立即交出執行權
                condLock.wait()    # 交出執行權，並等待另一個線程通知加偶數
            else:
                # 添奇數
                condLock.wait()  # 交出執行權，等待另一個線程通知加偶數
                lst.append(i)    
                condLock.notify()
        condLock.notify()


def odd():
    """加奇數"""
    with condLock:
        for i in range(1, 101, 2):
            if len(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)

事件鎖的應用

有 2 個任務線程來扮演李白和杜甫，如何讓他們一人一句進行對答？文本如下：

杜甫：老李啊，來喝酒！

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

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

杜甫：… 老李？

李白：呼呼呼… 睡着了…

代碼如下：

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

作者：雲崖君
https://www.cnblogs.com/YunyaSir/p/14963682.html

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/ETtU1eeIocLdoiw5wowuXQ

線程安全

鎖的作用

1、Lock() 同步鎖

基本介紹

使用方式

死鎖現象

with 語句

2、RLock() 遞歸鎖

基本介紹

使用方式

with 語句

3、Condition() 條件鎖

基本介紹

使用方式

with 語句

4、Event() 事件鎖

基本介紹

使用方式

5、Semaphore() 信號量鎖

基本介紹

使用方式

with 語句

鎖關係淺析

基本練習題

條件鎖的應用

事件鎖的應用

猜你喜歡