Python多进程线程读写锁实现方法

本文深入探讨了在Python多进程或多线程环境中，如何高效实现一个允许多个读者并发访问共享资源，同时确保单个作者独占且拥有优先权的读写锁机制。通过自定义`RWLock`类，利用`multiprocessing.JoinableQueue`进行进程间同步与通信，解决了传统锁机制在读写并发场景下的局限性，并提供了详细的代码示例和多线程适配方案，旨在帮助开发者构建更健壮、高性能的并发应用。

引言：并发读写共享资源的挑战

在多进程或多线程编程中，管理对共享资源的访问是一个核心问题。当存在一个或多个写入者（Writer）和多个读取者（Reader）同时操作一个共享文件或内存数据时，需要一套严谨的同步机制来保证数据的一致性和程序的正确性。理想的场景是：

1. 读操作并发性：允许多个读取者同时访问共享资源，以提高效率。
2. 写操作独占性：当写入者需要修改资源时，必须独占访问，防止其他读者读取到不一致的数据，也防止其他写入者同时修改。
3. 写入者优先权：当写入者请求访问时，应尽可能快地获得独占权，即使有读者正在进行读取。

传统的互斥锁（如multiprocessing.Lock或threading.Lock）可以保证独占访问，但它会限制所有操作的并发性，即使是读操作也无法并行。而multiprocessing.Condition虽然可以用于进程间的通知，但其设计更侧重于事件通知和等待，难以直接实现多读者并发、单作者独占且有优先权的复杂读写锁逻辑。

核心机制：自定义读写锁 RWLock

为了解决上述挑战，我们可以设计一个自定义的读写锁 RWLock。本方案的核心思想是利用 multiprocessing.JoinableQueue 作为读者与作者之间通信和同步的桥梁，实现一种合作式的读写控制。

RWLock 类设计理念

RWLock 的设计遵循以下原则：

• 读者等待作者：读者在开始读取前，会等待作者发出“新数据已准备好”的信号。
• 作者等待读者：作者在开始写入前，会等待所有读者完成对上一轮数据的读取。
• 作者优先权：作者可以通过设置一个“停止”标志，请求正在读取的读者尽快结束当前操作并释放资源，以便作者能快速获得独占权。

RWLock 类详解

以下是 RWLock 类的详细实现，适用于多进程环境：

from multiprocessing import Process, Lock, Value, JoinableQueue
from threading import local
import time

class RWLock:
    def __init__(self, num_readers: int):
        """
        创建一个支持单写入者和多个读取者的读写锁。
        num_readers: 预期的读取者数量。
        """
        if num_readers < 1 or not isinstance(num_readers, int):
            raise ValueError('num_readers 必须是一个正整数。')

        # _local_storage 用于为每个读者进程/线程分配一个唯一的队列
        self._local_storage = local() 
        self._num_readers = num_readers
        # _queue_count 用于在初始化时为读者分配队列
        self._queue_count = Value('i', 0) 
        # _stop 标志，作者用它来通知读者尽快停止读取
        self._stop = Value('i', 0) 
        # _lock 用于保护 _queue_count 的并发访问
        self._lock = Lock() 
        # 为每个读者创建一个 JoinableQueue
        self._queues = [JoinableQueue(1) for _ in range(self._num_readers)]

    def acquire_for_reading(self) -> None:
        """读者请求共享读取权限。"""
        # 为当前读者进程/线程分配一个队列（如果尚未分配）
        queue = getattr(self._local_storage, 'queue', None)
        if queue is None:
            with self._lock:
                # 使用 _queue_count 确保每个读者获得唯一的队列
                queue = self._queues[self._queue_count.value]
                self._queue_count.value += 1
            self._local_storage.queue = queue

        # 阻塞等待作者放入一个信号，表示有新数据可读
        queue.get() 

    def release_for_reading(self):
        """读者完成共享读取，释放权限。"""
        # 通知队列，读者已完成对当前数据的处理
        self._local_storage.queue.task_done()

    def acquire_for_writing(self, immediate=True):
        """
        作者请求独占写入权限。
        immediate: 如果为True，作者会设置停止标志，请求读者尽快释放。
        """
        if immediate:
            # 设置停止标志，通知读者尽快中断读取
            self._stop.value = 1 

        # 阻塞等待所有读者完成对上一轮数据的处理（通过 task_done）
        # 队列的 join() 方法会等待所有 put() 的任务被 task_done() 标记完成
        for queue in self._queues:
            queue.join()

    def release_for_writing(self) -> None:
        """作者完成独占写入，释放权限。"""
        # 重置停止标志
        self._stop.value = 0  

        # 向所有读者队列放入一个信号，唤醒等待的读者
        for queue in self._queues:
            queue.put(None)

    def is_stop_posted(self) -> bool:
        """
        读者周期性调用此函数，检查作者是否需要立即独占资源。
        """
        return True if self._stop.value else False

工作流程解析

1. 初始化 (__init__)：

   • 为每个预期读者创建一个 JoinableQueue。这些队列初始为空。
   • _queue_count 用于确保每个读者进程/线程在首次调用 acquire_for_reading 时能获得一个唯一的队列。
   • _stop 是一个共享的 Value，作为作者请求读者立即停止的标志。

2. 读者获取锁 (acquire_for_reading)：

   • 每个读者首次调用时，会从 _queues 列表中分配一个专属的 JoinableQueue。
   • 读者调用 queue.get()，由于队列初始为空，读者进程会在此处阻塞，等待作者放入数据。

3. 读者释放锁 (release_for_reading)：

   • 读者完成读取后，调用 queue.task_done()。这会通知其专属队列，之前由作者放入的那个任务（即唤醒信号）已经处理完毕。

4. 作者获取锁 (acquire_for_writing)：

   • 作者遍历所有读者的 JoinableQueue，并调用 queue.join()。
   • queue.join() 会阻塞，直到队列中所有通过 put() 添加的任务都被 task_done() 标记为完成。这意味着作者会等待所有读者完成对上一轮数据的读取。
   • 如果 immediate 参数为 True，作者会设置 _stop.value = 1，向读者发出立即停止的信号。

5. 作者释放锁 (release_for_writing)：

   • 作者完成写入后，首先重置 _stop.value = 0。
   • 然后，作者向每个读者的 JoinableQueue 中 put(None)。这会解除之前 acquire_for_reading 中读者 queue.get() 的阻塞状态，允许读者开始读取新数据。

6. 作者优先权 (is_stop_posted)：

   • 为了实现作者的“立即”优先权，读者在进行长时间读取操作时，需要周期性地调用 is_stop_posted()。如果检测到 _stop 标志被设置，读者应立即中断当前读取并调用 release_for_reading() 释放锁。

示例代码：多进程读写场景

以下是一个使用 RWLock 实现多进程读写共享数据的完整示例：

from multiprocessing import Process, Lock, Value, JoinableQueue
from threading import local
import time

# RWLock 类定义如上所示...
class RWLock:
    def __init__(self, num_readers: int):
        if num_readers < 1 or not isinstance(num_readers, int):
            raise ValueError('num_readers 必须是一个正整数。')
        self._local_storage = local()
        self._num_readers = num_readers
        self._queue_count = Value('i', 0)
        self._stop = Value('i', 0)
        self._lock = Lock()
        self._queues = [JoinableQueue(1) for _ in range(self._num_readers)]

    def acquire_for_reading(self) -> None:
        queue = getattr(self._local_storage, 'queue', None)
        if queue is None:
            with self._lock:
                queue = self._queues[self._queue_count.value]
                self._queue_count.value += 1
            self._local_storage.queue = queue
        queue.get()

    def release_for_reading(self):
        self._local_storage.queue.task_done()

    def acquire_for_writing(self, immediate=True):
        if immediate:
            self._stop.value = 1;

        for queue in self._queues:
            queue.join()

    def release_for_writing(self) -> None:
        self._stop.value = 0

        for queue in self._queues:
            queue.put(None)

    def is_stop_posted(self) -> bool:
        return True if self._stop.value else False

# 共享数据类，使用 multiprocessing.Value 实现进程间共享
class SharedData:
    def __init__(self):
        self.value = Value('i', 0, lock=False) # lock=False 表示不使用内部锁，由 RWLock 管理

def reader(rw_lock, id, shared_data):
    while True:
        rw_lock.acquire_for_reading()
        # 模拟长时间读取任务
        # 在读取过程中周期性检查作者是否需要独占
        sleep_time = id / 10 # 不同的读者有不同的模拟读取时间
        for _ in range(10):
            time.sleep(sleep_time)
            if rw_lock.is_stop_posted():
                print(f'读者 {id} 收到停止信号，中断读取。', flush=True)
                break # 中断当前读取循环

        print(f'读者 {id} 完成处理数据: {shared_data.value}', flush=True)
        rw_lock.release_for_reading()
        time.sleep(0.1) # 短暂休眠，避免忙等待

def writer(rw_lock, shared_data):
    while True:
        # 当 shared_data.value 为 3 时，作者请求立即独占
        rw_lock.acquire_for_writing(immediate=(shared_data.value == 3))
        shared_data.value.value += 1 # 修改共享数据
        print(f'作者写入数据: {shared_data.value.value} (时间: {time.time()})', flush=True)
        rw_lock.release_for_writing()
        time.sleep(0.5) # 模拟作者完成写入后的其他工作

def main():
    rw_lock = RWLock(3) # 实例化读写锁，有3个读者
    shared_data = SharedData() # 共享数据

    # 启动读者进程
    for id in range(1, 4):
        Process(target=reader, args=(rw_lock, id, shared_data), daemon=True).start()

    # 启动作者进程
    Process(target=writer, args=(rw_lock, shared_data), daemon=True).start()

    input('按 Enter 键终止程序:\n')

if __name__ == '__main__':
    main()

运行上述代码，你将观察到：

• 作者会写入数据，然后读者会并发地读取这些数据。
• 读者会根据其 id 模拟不同的读取时间。
• 当 shared_data.value 达到 3 时，作者会以 immediate=True 的方式获取锁。此时，正在读取的读者会检查 is_stop_posted() 标志，如果发现被设置，会尽快中断读取并释放锁，从而让作者更快地获得独占权。

多线程适配

上述 RWLock 类是为 multiprocessing 设计的，但其核心逻辑同样适用于多线程环境。只需要将 multiprocessing 相关的原语替换为 threading 和 queue 模块中的对应实现即可。

主要替换点：

• multiprocessing.Process -> threading.Thread
• multiprocessing.Lock -> threading.Lock
• multiprocessing.Value -> 普通 Python int 变量（因为线程共享内存，不需要特殊的共享值类型）
• multiprocessing.JoinableQueue -> queue.Queue

以下是多线程版本的 RWLockMultiThreading 实现：

from threading import Thread, Lock, local
from queue import Queue
import time

class RWLockMultiThreading:
    def __init__(self, num_readers: int):
        """
        创建一个支持单写入者和多个读取者（多线程）的读写锁。
        """
        if num_readers < 1 or not isinstance(num_readers, int):
            raise ValueError('num_readers 必须是一个正整数。')

        self._local_storage = local()
        self._num_readers = num_readers
        self._queue_count = 0 # 普通int变量，因为线程共享内存
        self._stop = 0        # 普通int变量
        self._lock = Lock()   # threading.Lock
        self._queues = [Queue(1) for _ in range(self._num_readers)] # queue.Queue

    def acquire_for_reading(self) -> None:
        queue = getattr(self._local_storage, 'queue', None)
        if queue is None:
            with self._lock:
                queue = self._queues[self._queue_count]
                self._queue_count += 1
            self._local_storage.queue = queue
        queue.get()

    def release_for_reading(self):
        self._local_storage.queue.task_done()

    def acquire_for_writing(self, immediate=True):
        if immediate:
            self._stop = 1;

        for queue in self._queues:
            queue.join()

    def release_for_writing(self) -> None:
        self._stop = 0

        for queue in self._queues:
            queue.put(None)

    def is_stop_posted(self) -> bool:
        return True if self._stop else False

# 共享数据类，普通Python对象即可
class SharedValue:
    def __init__(self):
        self.value = 0

def reader_thread(rw_lock, id, shared_data):
    while True:
        rw_lock.acquire_for_reading()
        sleep_time = id / 10
        for _ in range(10):
            time.sleep(sleep_time)
            if rw_lock.is_stop_posted():
                print(f'读者线程 {id} 收到停止信号，中断读取。', flush=True)
                break
        print(f'读者线程 {id} 完成处理数据: {shared_data.value}', flush=True)
        rw_lock.release_for_reading()
        time.sleep(0.1)

def writer_thread(rw_lock, shared_data):
    while True:
        rw_lock.acquire_for_writing(immediate=(shared_data.value == 3))
        shared_data.value += 1
        print(f'作者线程写入数据: {shared_data.value} (时间: {time.time()})', flush=True)
        rw_lock.release_for_writing()
        time.sleep(0.5)

def main_thread():
    rw_lock = RWLockMultiThreading(3)
    shared_data = SharedValue()
    for id in range(1, 4):
        Thread(target=reader_thread, args=(rw_lock, id, shared_data), daemon=True).start()
    Thread(target=writer_thread, args=(rw_lock, shared_data), daemon=True).start()
    input('按 Enter 键终止程序:\n')

if __name__ == '__main__':
    main_thread()

注意事项与最佳实践

1. 读者协作性：作者的“立即”优先权依赖于读者进程/线程的协作。如果读者在执行长时间任务时不检查 is_stop_posted() 标志，作者将无法强制其立即释放资源，只能等待读者自然完成。因此，在实际应用中，读者任务应设计成可中断的，并在关键点检查此标志。
2. 数据一致性：此 RWLock 确保了写入时的独占性，即当作者写入时，没有读者正在读取，也没有其他作者（本设计只支持一个作者）正在写入。同时，它也保证了读者总是读取到作者完整写入后的数据，避免了读取到不一致的中间状态。
3. 资源管理：务必确保 acquire_for_reading() 和 acquire_for_writing() 之后，都对应地调用 release_for_reading() 和 release_for_writing()，即使在异常情况下也应如此（例如使用 try...finally 结构），以避免死锁或资源泄露。 4