MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Python锁机制在多线程中的应用实例

2022-07-273.3k 阅读

多线程编程中的资源竞争问题

在多线程编程中，多个线程可能同时访问和修改共享资源，这就可能导致资源竞争问题。例如，考虑一个简单的银行账户类，其中有一个余额属性，当多个线程同时进行存款或取款操作时，如果不加以控制，就可能出现数据不一致的情况。

假设有如下Python代码：

import threading


class BankAccount:
    def __init__(self):
        self.balance = 0


def deposit(account, amount):
    account.balance += amount


def withdraw(account, amount):
    if account.balance >= amount:
        account.balance -= amount


account = BankAccount()
threads = []
for _ in range(10):
    t1 = threading.Thread(target=deposit, args=(account, 100))
    t2 = threading.Thread(target=withdraw, args=(account, 50))
    threads.append(t1)
    threads.append(t2)

for t in threads:
    t.start()

for t in threads:
    t.join()

print(f"Final balance: {account.balance}")

在这段代码中，BankAccount类有一个balance属性表示账户余额。deposit函数用于存款，withdraw函数用于取款。创建了多个线程对账户进行存款和取款操作。然而，由于多个线程同时访问和修改balance属性，会出现资源竞争问题，导致最终的余额可能不符合预期。这是因为线程切换可能发生在if account.balance >= amount判断之后，但在account.balance -= amount执行之前，这样就可能取出超过账户余额的钱。

Python锁机制概述

为了解决多线程编程中的资源竞争问题，Python提供了锁机制。锁（Lock）是一种同步原语，它只有两种状态：锁定（locked）和未锁定（unlocked）。线程在访问共享资源之前，必须先获取锁（将锁的状态从未锁定变为锁定），访问完成后释放锁（将锁的状态从锁定变为未锁定）。其他线程在锁处于锁定状态时，无法获取锁，只能等待，直到锁被释放。

在Python的threading模块中，Lock类用于创建锁对象。其基本使用方法如下：

import threading


lock = threading.Lock()
# 获取锁
lock.acquire()
try:
    # 访问共享资源的代码
    pass
finally:
    # 释放锁
    lock.release()

在上述代码中，使用lock.acquire()获取锁，如果锁当前未被锁定，则获取成功，线程可以继续执行；如果锁已被锁定，则线程会阻塞，直到锁被释放。使用try - finally块来确保无论在访问共享资源的过程中是否发生异常，锁都会被释放，避免死锁。

Python锁机制在多线程中的应用实例

简单计数器示例

下面通过一个简单的计数器示例，展示锁机制如何解决多线程资源竞争问题。假设有多个线程对一个计数器进行递增操作：

import threading


class Counter:
    def __init__(self):
        self.value = 0
        self.lock = threading.Lock()


def increment(counter):
    for _ in range(10000):
        counter.lock.acquire()
        try:
            counter.value += 1
        finally:
            counter.lock.release()


counter = Counter()
threads = []
for _ in range(10):
    t = threading.Thread(target=increment, args=(counter,))
    threads.append(t)

for t in threads:
    t.start()

for t in threads:
    t.join()

print(f"Final counter value: {counter.value}")

在Counter类中，除了value属性外，还创建了一个lock锁对象。在increment函数中，每次对counter.value进行递增操作前，先获取锁，操作完成后释放锁。这样，当多个线程同时执行increment函数时，由于锁的存在，同一时间只有一个线程能够修改counter.value，从而避免了资源竞争问题，确保最终的计数器值是正确的（预期为100000）。

复杂数据结构示例 - 共享字典

再来看一个更复杂的示例，使用共享字典作为共享资源。假设有多个线程向共享字典中添加键值对，并且可能会查询字典中的值。

import threading


class SharedDict:
    def __init__(self):
        self.data = {}
        self.lock = threading.Lock()


def add_item(shared_dict, key, value):
    shared_dict.lock.acquire()
    try:
        shared_dict.data[key] = value
    finally:
        shared_dict.lock.release()


def get_item(shared_dict, key):
    shared_dict.lock.acquire()
    try:
        return shared_dict.data.get(key)
    finally:
        shared_dict.lock.release()


shared_dict = SharedDict()
threads = []
for i in range(5):
    t1 = threading.Thread(target=add_item, args=(shared_dict, f"key_{i}", i))
    t2 = threading.Thread(target=get_item, args=(shared_dict, f"key_{i}"))
    threads.append(t1)
    threads.append(t2)

for t in threads:
    t.start()

for t in threads:
    t.join()

在这个示例中，SharedDict类包含一个data字典和一个lock锁。add_item函数用于向字典中添加键值对，get_item函数用于从字典中获取值。在这两个函数中，都通过获取锁来确保对data字典的操作是线程安全的。这样，即使多个线程同时进行添加和查询操作，也不会出现数据不一致的问题。

银行账户示例改进

回到前面的银行账户示例，使用锁机制来改进代码，使其能够正确处理多线程下的存款和取款操作。

import threading


class BankAccount:
    def __init__(self):
        self.balance = 0
        self.lock = threading.Lock()


def deposit(account, amount):
    account.lock.acquire()
    try:
        account.balance += amount
    finally:
        account.lock.release()


def withdraw(account, amount):
    account.lock.acquire()
    try:
        if account.balance >= amount:
            account.balance -= amount
    finally:
        account.lock.release()


account = BankAccount()
threads = []
for _ in range(10):
    t1 = threading.Thread(target=deposit, args=(account, 100))
    t2 = threading.Thread(target=withdraw, args=(account, 50))
    threads.append(t1)
    threads.append(t2)

for t in threads:
    t.start()

for t in threads:
    t.join()

print(f"Final balance: {account.balance}")

在改进后的代码中，BankAccount类添加了一个lock锁。deposit和withdraw函数在访问和修改balance属性前，都先获取锁，操作完成后释放锁。这样就保证了在多线程环境下，存款和取款操作的原子性，避免了资源竞争导致的余额错误问题。

锁机制的深入理解

死锁问题

虽然锁机制能够有效解决多线程资源竞争问题，但如果使用不当，可能会导致死锁。死锁是指两个或多个线程相互等待对方释放锁，从而导致所有线程都无法继续执行的情况。例如：

import threading


lock1 = threading.Lock()
lock2 = threading.Lock()


def thread1():
    lock1.acquire()
    print("Thread 1 acquired lock1")
    lock2.acquire()
    print("Thread 1 acquired lock2")
    lock2.release()
    lock1.release()


def thread2():
    lock2.acquire()
    print("Thread 2 acquired lock2")
    lock1.acquire()
    print("Thread 2 acquired lock1")
    lock1.release()
    lock2.release()


t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)

t1.start()
t2.start()

t1.join()
t2.join()

在上述代码中，thread1先获取lock1，然后尝试获取lock2；而thread2先获取lock2，然后尝试获取lock1。如果thread1获取了lock1，thread2获取了lock2，此时两个线程就会相互等待对方释放锁，从而导致死锁。

为了避免死锁，可以采取以下几种方法：

  1. 按顺序获取锁：所有线程都按照相同的顺序获取锁。例如，如果有多个锁lock1、lock2、lock3，所有线程都先获取lock1，再获取lock2，最后获取lock3。这样就不会出现相互等待的情况。
  2. 使用超时机制：在获取锁时设置一个超时时间。如果在指定时间内未能获取到锁，线程可以释放已经获取的锁，并进行其他操作，避免无限期等待。例如：
import threading
import time


lock1 = threading.Lock()
lock2 = threading.Lock()


def thread1():
    if lock1.acquire(timeout=1):
        print("Thread 1 acquired lock1")
        time.sleep(0.5)
        if lock2.acquire(timeout=1):
            print("Thread 1 acquired lock2")
            lock2.release()
        lock1.release()


def thread2():
    if lock2.acquire(timeout=1):
        print("Thread 2 acquired lock2")
        time.sleep(0.5)
        if lock1.acquire(timeout=1):
            print("Thread 2 acquired lock1")
            lock1.release()
        lock2.release()


t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)

t1.start()
t2.start()

t1.join()
t2.join()

在这个示例中，acquire方法设置了timeout参数为1秒。如果在1秒内未能获取到锁，线程会放弃获取，并继续执行后续代码，从而避免死锁。

锁的性能开销

虽然锁机制能够保证多线程编程的正确性，但它也带来了一定的性能开销。每次获取和释放锁都需要一定的时间，尤其是在高并发环境下，频繁的锁操作可能会成为性能瓶颈。例如，在一个循环中对共享资源进行大量的简单操作，如果每次操作都获取和释放锁，会导致锁竞争过于频繁，降低程序的执行效率。

为了优化性能，可以考虑以下几种方法：

  1. 减少锁的粒度：尽量缩小锁保护的代码块范围，只对真正需要保护的共享资源操作加锁。例如，在一个包含多个操作的函数中，如果只有部分操作涉及共享资源，只对这部分操作加锁，而不是对整个函数加锁。
  2. 使用读写锁：如果共享资源的读操作远多于写操作，可以使用读写锁（RLock）。读写锁允许多个线程同时进行读操作，但只允许一个线程进行写操作。这样可以提高读操作的并发性能。例如：
import threading


class SharedData:
    def __init__(self):
        self.data = 0
        self.lock = threading.RLock()


def read_data(shared_data):
    shared_data.lock.acquire()
    try:
        print(f"Read data: {shared_data.data}")
    finally:
        shared_data.lock.release()


def write_data(shared_data, value):
    shared_data.lock.acquire()
    try:
        shared_data.data = value
        print(f"Write data: {shared_data.data}")
    finally:
        shared_data.lock.release()


shared_data = SharedData()
read_threads = []
write_thread = threading.Thread(target=write_data, args=(shared_data, 100))
for _ in range(5):
    t = threading.Thread(target=read_data, args=(shared_data,))
    read_threads.append(t)

write_thread.start()
for t in read_threads:
    t.start()

write_thread.join()
for t in read_threads:
    t.join()

在这个示例中，SharedData类使用RLock作为锁。read_data函数用于读操作，write_data函数用于写操作。多个读线程可以同时获取锁进行读操作，而写线程获取锁时会独占锁，其他读线程和写线程都需要等待。

总结

Python的锁机制是解决多线程编程中资源竞争问题的重要手段。通过合理使用锁，可以确保多个线程安全地访问共享资源，避免数据不一致和其他并发问题。然而，在使用锁的过程中，需要注意死锁问题和性能开销。通过按顺序获取锁、使用超时机制、减少锁的粒度以及使用读写锁等方法，可以有效地避免死锁并提高程序的性能。在实际的多线程编程中，需要根据具体的应用场景和需求，灵活运用锁机制，以实现高效、安全的多线程程序。同时，除了锁机制外，Python还提供了其他同步原语，如信号量（Semaphore）、条件变量（Condition）等，这些同步原语在不同的场景下也有各自的用途，可以进一步丰富多线程编程的手段。