面试题：Python多线程中如何避免资源竞争问题

Question

在Python多线程编程时，经常会遇到多个线程同时访问共享资源的情况，从而引发资源竞争。请描述你所知道的避免资源竞争的方法，并举例说明如何使用Python的`threading`模块中的工具来实现这些方法。

Answer 1

避免资源竞争的方法

1. 锁（Lock）：
   • 原理：锁是一种简单的同步原语，一个线程获取锁后，其他线程若想获取锁就会被阻塞，直到该线程释放锁。这样就保证了同一时间只有一个线程能访问共享资源。
   • 应用场景：适用于对共享资源的简单读写操作，保证同一时刻只有一个线程能访问。
2. 信号量（Semaphore）：
   • 原理：信号量维护了一个内部计数器，每次获取信号量时计数器减1，释放信号量时计数器加1。当计数器为0时，获取信号量的操作会被阻塞，直到有其他线程释放信号量。
   • 应用场景：适用于限制同时访问共享资源的线程数量。比如数据库连接池，限制同时连接数据库的线程数。
3. 条件变量（Condition）：
   • 原理：条件变量允许线程在满足特定条件时才进行相应操作。线程可以等待条件变量，也可以通知其他等待的线程条件已经满足。
   • 应用场景：适用于线程间需要基于某个条件进行协作的场景。例如生产者 - 消费者模型，生产者生产数据后通知消费者，消费者在有数据时才进行消费。
4. 队列（Queue）：
   • 原理：队列本身是线程安全的。生产者线程将数据放入队列，消费者线程从队列中取出数据，队列内部的机制保证了在多线程环境下数据的正确处理，避免资源竞争。
   • 应用场景：常用于生产者 - 消费者模型，解耦生产者和消费者线程，让它们以不同的速率工作。

使用threading模块实现

1. 锁（Lock）示例：

import threading

# 共享资源
counter = 0
lock = threading.Lock()


def increment():
    global counter
    for _ in range(1000000):
        # 获取锁
        lock.acquire()
        try:
            counter += 1
        finally:
            # 释放锁
            lock.release()


threads = []
for _ in range(2):
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

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

2. 信号量（Semaphore）示例：

import threading

# 模拟数据库连接池，最多允许3个线程同时连接
semaphore = threading.Semaphore(3)


def access_database():
    semaphore.acquire()
    try:
        print(f"{threading.current_thread().name} accessing database")
        # 模拟数据库操作
        import time
        time.sleep(2)
        print(f"{threading.current_thread().name} finished accessing database")
    finally:
        semaphore.release()


threads = []
for _ in range(5):
    t = threading.Thread(target=access_database)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

3. 条件变量（Condition）示例：

import threading

condition = threading.Condition()
data_ready = False


def producer():
    global data_ready
    with condition:
        print("Producer is producing data...")
        data_ready = True
        condition.notify()


def consumer():
    with condition:
        print("Consumer is waiting for data...")
        condition.wait_for(lambda: data_ready)
        print("Consumer got the data and is processing it...")


producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

consumer_thread.start()
producer_thread.start()

producer_thread.join()
consumer_thread.join()

4. 队列（Queue）示例：

import threading
import queue

q = queue.Queue()


def producer():
    for i in range(5):
        q.put(i)
        print(f"Producer added {i} to the queue")


def consumer():
    while True:
        item = q.get()
        if item is None:
            break
        print(f"Consumer got {item} from the queue")
        q.task_done()


producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

producer_thread.join()
q.put(None)
consumer_thread.join()