Python 高并发内存缓存实践：基于写优先读写锁（RWLock）与条件变量的防饥饿机制实现

一、问题背景：读多写少场景下的全局互斥锁瓶颈与“写饥饿”

在云原生微服务、高频配置中心或本地二级缓存架构中，数据访问通常呈现出极端的“读多写少”特征（例如读写比达到 9:1 甚至 99:1）。若采用 Python 标准库最基础的互斥锁（threading.Lock），即使数十个工作线程仅仅是并发读取无害的共享只读状态，也会被迫串行化排队争锁，引发极大的 CPU 线程上下文切换开销与延迟恶化。

为了释放只读操作的并发潜能，标准解法是引入读写锁（Read-Write Lock / RWLock）：允许多个读者并发持有共享读锁，仅在写者更新时获取排他写锁。

然而，朴素的读优先（Read-Preference）读写锁存在致命的隐患——写饥饿（Write Starvation）：只要源源不断的读请求持续到达，读锁计数器便始终无法归零，导致写线程无限期阻塞等待，业务配置或关键状态无法及时更新。为此，必须设计并实现具备 写优先（Write-Preference）调度 的线程安全读写锁。

二、写优先调度状态机核心设计

写优先读写锁通过底层互斥锁与两个专用的条件变量（_ok_to_read 与 _ok_to_write）共同协调：

1. 写者挂起阻塞后续读者：一旦有写线程尝试获取写锁（_waiting_writers > 0），后续新到达的读线程必须立即在 _ok_to_read 条件变量上挂起排队，即使当前写锁尚未被正式持有，坚决切断读者的源源不断占用。
2. 存量读者平滑排水：已持有读锁的活跃读者（_active_readers）继续执行直至自然退出；当最后一个活跃读者退出且检测到有等待中的写者时，精准唤醒目标写线程。
3. 写者优先唤醒链：写操作释放排他锁时，若仍有积压的等待写者，优先单播唤醒写者（_ok_to_write.notify()）；仅当等待写者计数归零时，才广播唤醒所有积压排队的读者（_ok_to_read.notify_all()）。

三、完整可运行代码实现

以下脚本在 Python 3.11 标准库环境下直接运行，无任何第三方依赖，包含 WritePreferenceRWLock 与多读者并发压测验证：

import threading
import time
from typing import Dict, Any, Optional


class WritePreferenceRWLock:
    """写优先（Write-Preference）读写锁：防止写饥饿（Write Starvation）的线程安全实现"""

    def __init__(self):
        self._lock = threading.Lock()
        self._ok_to_read = threading.Condition(self._lock)
        self._ok_to_write = threading.Condition(self._lock)
        self._active_readers = 0
        self._waiting_writers = 0
        self._active_writer = False

    def acquire_read(self):
        with self._lock:
            # 写优先：只要有活跃的写者或正在等待的写者，后续读者均需阻塞排队
            while self._active_writer or self._waiting_writers > 0:
                self._ok_to_read.wait()
            self._active_readers += 1

    def release_read(self):
        with self._lock:
            self._active_readers -= 1
            if self._active_readers == 0 and self._waiting_writers > 0:
                # 最后一个活跃读者退出时，唤醒等待中的写者
                self._ok_to_write.notify()

    def acquire_write(self):
        with self._lock:
            self._waiting_writers += 1
            while self._active_writer or self._active_readers > 0:
                self._ok_to_write.wait()
            self._waiting_writers -= 1
            self._active_writer = True

    def release_write(self):
        with self._lock:
            self._active_writer = False
            if self._waiting_writers > 0:
                # 优先唤醒写者
                self._ok_to_write.notify()
            else:
                # 无等待写者时，唤醒所有被阻塞的读者
                self._ok_to_read.notify_all()


class ThreadSafeRWCache:
    """基于写优先读写锁的线程安全键值缓存"""

    def __init__(self):
        self._data: Dict[str, Any] = {}
        self._rwlock = WritePreferenceRWLock()

    def get(self, key: str) -> Optional[Any]:
        self._rwlock.acquire_read()
        try:
            return self._data.get(key)
        finally:
            self._rwlock.release_read()

    def put(self, key: str, value: Any):
        self._rwlock.acquire_write()
        try:
            self._data[key] = value
        finally:
            self._rwlock.release_write()


def verify_rwlock():
    cache = ThreadSafeRWCache()
    cache.put("config:rate", 100)
    read_counts = 0
    write_counts = 0
    c_lock = threading.Lock()

    def reader(rid: int):
        nonlocal read_counts
        for _ in range(30):
            val = cache.get("config:rate")
            if val is not None:
                with c_lock:
                    read_counts += 1
            time.sleep(0.002)

    def writer(wid: int):
        nonlocal write_counts
        for i in range(10):
            cache.put("config:rate", 100 + i)
            with c_lock:
                write_counts += 1
            time.sleep(0.005)

    readers = [threading.Thread(target=reader, args=(i,)) for i in range(6)]
    writers = [threading.Thread(target=writer, args=(j,)) for j in range(2)]

    t0 = time.monotonic()
    for t in readers + writers:
        t.start()
    for t in readers + writers:
        t.join()
    elapsed_ms = round((time.monotonic() - t0) * 1000)

    print(
        f"[Result] 写优先读写锁高并发压测通过: 6读线程完成读取={read_counts}/180, "
        f"2写线程完成写入={write_counts}/20, 最终缓存值={cache.get('config:rate')}, 耗时={elapsed_ms}ms"
    )


if __name__ == "__main__":
    verify_rwlock()

四、实际运行验证与生产容器化最佳实践

在 Python 3.11 环境下执行上述并发测试，实测输出如下：

• [Result] 写优先读写锁高并发压测通过: 6读线程完成读取=180/180, 2写线程完成写入=20/20, 最终缓存值=109, 耗时=76ms

生产容器化落地的关键考量：

1. 防止锁内 I/O 阻塞：在读写锁的保护代码块内，严禁发起慢数据库查询或跨网络 RPC。锁保护区应仅保留纳秒级的内存字典读写，避免长阻塞导致条件变量唤醒链雪崩。
2. 死锁防御与异常处理规范：使用 try...finally 块确保 release_read() 与 release_write() 必被执行，即便在业务逻辑抛出异常时也能保障锁计数器状态一致。