代码收藏家技术教程 2025-01-07

Python：线程锁

在 Python 中，多线程编程通常是通过 `threading` 模块实现的，线程锁是这个模块中一个非常重要的概念。线程锁帮助开发者控制对共享资源的安全访问，从而避免潜在的竞争条件（race conditions）和数据不一致性。以下是对 Python 中线程锁的详细介绍。

1. 互斥锁（Mutex）

互斥锁是 Python 中最常用的锁类型，保证在任何时刻只能有一个线程访问某个资源。`threading.Lock` 提供了互斥锁的实现。

1.1 使用方法

        创建锁：使用 `threading.Lock()` 创建一个锁实例。
        获取锁：使用 `lock.acquire()` 方法获取锁。
        释放锁：使用 `lock.release()` 方法释放锁。
        上下文管理：可以使用 `with` 语句来自动管理锁的获取和释放。

1.2 示例代码：

import threading

num = 0
# 创建锁
mutex = threading.Lock()

def num1_fun():
    # 上锁
    mutex.acquire()
    global num
    for i in range(10000000):
        num += 1
    # 解锁
    mutex.release()
    print(f'num1:{num}')


def num2_fun():
    # 上锁
    mutex.acquire()
    global num
    for i in range(10000000):
        num += 1
    # 解锁
    mutex.release()
    print(f'num2:{num}')


if __name__ == '__main__':
    # 创建线程
    num1_thread = threading.Thread(target=num1_fun)
    num2_thread = threading.Thread(target=num2_fun)
    # 启动线程
    num1_thread.start()
    num2_thread.start()

'''
    未加锁结果：
    num1:12491225
    num2:12589772
    加锁之后的值：
    num1:10000000
    num2:20000000
'''

2. 递归锁（Recursive Lock）

递归锁是允许同一线程多次获取锁的锁类型。Python 中的递归锁实现为 `threading.RLock`。

2.1 使用方法

和 `Lock` 类似，但可以在同一线程中重复调用 `acquire()` 方法而不会造成死锁。
每次调用 `acquire()`，需要呼叫相应数量的 `release()` 来释放锁。

2.2 示例代码

import threading  

# 创建一个递归锁  
rlock = threading.RLock()  

def recursive_function(n):  
    if n > 0:  
        rlock.acquire()  
        print(f"Recursion depth: {n}")  
        recursive_function(n - 1)  # 递归调用  
        rlock.release()  

# 启动线程  
thread = threading.Thread(target=recursive_function, args=(5,))  
thread.start()  
thread.join()

3. 条件变量（Condition Variable）

条件变量用于让线程在特定条件下进行等待和唤醒配合。通常与互斥锁一起使用。

3.1 使用方法

        创建条件变量：使用 `threading.Condition()`。
        在线程中使用 `condition.wait()` 使线程进入等待状态。
        在其他线程中使用 `condition.notify()` 或 `condition.notify_all()` 唤醒等待的线程。

3.2 示例代码

import threading  
import time  

condition = threading.Condition()  
data_ready = False  

def producer():  
    global data_ready  
    time.sleep(2)  # 模拟生产数据  
    with condition:  
        data_ready = True  
        condition.notify()  # 通知消费者  

def consumer():  
    with condition:  
        while not data_ready:  
            condition.wait()  # 等待生产者的通知  
        print("Data is ready to be consumed.")  

# 启动线程  
producer_thread = threading.Thread(target=producer)  
consumer_thread = threading.Thread(target=consumer)  

consumer_thread.start()  
producer_thread.start()  

consumer_thread.join()  
producer_thread.join()

4. 信号量（Semaphore）

信号量是一个计数锁，允许多个线程同时访问限制数量的资源。Python 中的 `threading.Semaphore` 实现了信号量。

4.1 使用方法

        使用 `threading.Semaphore(max_count)` 创建一个信号量，`max_count` 为最大允许的线程数。
        获取信号量：使用 `semaphore.acquire()`。
        释放信号量：使用 `semaphore.release()`。

4.2 示例代码：

import threading  
import time  

# 创建信号量，最多允许 2 个线程同时访问  
semaphore = threading.Semaphore(2)  

def access_resource(thread_id):  
    semaphore.acquire()  # 获取信号量  
    print(f"Thread {thread_id} has accessed the resource.")  
    time.sleep(2)  # 模拟访问资源  
    print(f"Thread {thread_id} is releasing the resource.")  
    semaphore.release()  # 释放信号量  

# 创建多个线程  
threads = []  
for i in range(5):  
    thread = threading.Thread(target=access_resource, args=(i,))  
    threads.append(thread)  
    thread.start()  

# 等待所有线程完成  
for thread in threads:  
    thread.join()