面试题：Python锁机制在高并发场景下的性能优化与死锁避免

锁的分配策略

粒度控制：
- 对于不同类型的资源，尽量使用细粒度锁。例如，对于文件句柄资源可以用一个锁，数据库连接资源用另一个锁，内存缓冲区资源用专门的锁。这样，不同线程在竞争不同资源时不会相互阻塞。
- 比如，在处理文件操作时：

import threading

file_lock = threading.Lock()

def write_to_file(file_path, content):
    with file_lock:
        with open(file_path, 'w') as f:
            f.write(content)

资源排序：
- 对所有需要锁的资源进行全局排序。在获取多个锁时，所有线程都按照这个固定顺序获取锁。例如，假设有资源A（文件句柄）、B（数据库连接）、C（内存缓冲区），都按照字母顺序获取锁。

resource_a_lock = threading.Lock()
resource_b_lock = threading.Lock()
resource_c_lock = threading.Lock()

def use_resources():
    with resource_a_lock:
        with resource_b_lock:
            with resource_c_lock:
                # 执行需要使用这三个资源的操作
                pass

死锁检测与恢复机制

死锁检测：
- 超时检测：在获取锁时设置超时时间。例如，当一个线程获取第一个锁后，在规定时间内无法获取第二个锁，则释放已获取的锁，重新尝试。

import time

def acquire_locks(locks, timeout=5):
    acquired = [False] * len(locks)
    start_time = time.time()
    for i, lock in enumerate(locks):
        while time.time() - start_time < timeout:
            if lock.acquire(blocking=False):
                acquired[i] = True
                break
            time.sleep(0.1)
        else:
            for j in range(i):
                if acquired[j]:
                    locks[j].release()
            return False
    return True

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

if acquire_locks([lock1, lock2]):
    try:
        # 执行操作
        pass
    finally:
        lock2.release()
        lock1.release()

- **资源分配图算法**：维护一个资源分配图，记录线程与资源的占用和请求关系。通过算法（如死锁定理）定期检查图中是否存在环，如果存在环则表示可能发生死锁。不过，Python 标准库中没有直接实现该算法，需要自行编写数据结构和算法逻辑来实现。

2. 死锁恢复： - 撤销线程：当检测到死锁时，选择撤销一个或多个线程，释放它们持有的锁，让其他线程能够继续执行。例如，可以记录每个线程的执行状态和资源占用情况，选择占用资源较少或优先级较低的线程进行撤销。 - 重试机制：被撤销的线程可以在一段时间后重新尝试获取锁并执行任务。

性能测试方法

时间测量：
- 使用 timeit 模块测量单个操作（如获取锁并执行相关资源操作）的时间。

import timeit

def test_write_to_file():
    write_to_file('test.txt', 'test content')

time_taken = timeit.timeit(test_write_to_file, number = 1000)
print(f"Time taken for 1000 runs: {time_taken} seconds")

多线程性能测试：
- 创建多个线程同时竞争资源，使用 threading 模块和 time 模块测量总运行时间。

import threading
import time

def worker():
    write_to_file('test.txt', 'worker content')

threads = []
start_time = time.time()
for _ in range(10):
    t = threading.Thread(target = worker)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

end_time = time.time()
print(f"Total time for 10 threads: {end_time - start_time} seconds")

对比测试：
- 对比不同锁分配策略（如粗粒度锁和细粒度锁）下的性能。通过分别实现不同策略，并使用上述性能测试方法，观察性能差异，选择最优策略。

星途面试题库

面试题：Python锁机制在高并发场景下的性能优化与死锁避免

知识考点

面试题答案

锁的分配策略

死锁检测与恢复机制

性能测试方法