本文深入探讨了在python多进程或多线程环境中，如何高效实现一个允许多个读者并发访问共享资源，同时确保单个作者独占且拥有优先权的读写锁机制。通过自定义`rwlock`类，利用`multiprocessing.joinablequeue`进行进程间同步与通信，解决了传统锁机制在读写并发场景下的局限性，并提供了详细的代码示例和多线程适配方案，旨在帮助开发者构建更健壮、高性能的并发应用。
引言：并发读写共享资源的挑战
在多进程或多线程编程中，管理对共享资源的访问是一个核心问题。当存在一个或多个写入者（Writer）和多个读取者（Reader）同时操作一个共享文件或内存数据时，需要一套严谨的同步机制来保证数据的一致性和程序的正确性。理想的场景是：
- 读操作并发性：允许多个读取者同时访问共享资源，以提高效率。
- 写操作独占性：当写入者需要修改资源时，必须独占访问，防止其他读者读取到不一致的数据，也防止其他写入者同时修改。
- 写入者优先权：当写入者请求访问时，应尽可能快地获得独占权，即使有读者正在进行读取。
传统的互斥锁（如multiprocessing.Lock或threading.Lock）可以保证独占访问，但它会限制所有操作的并发性，即使是读操作也无法并行。而multiprocessing.Condition虽然可以用于进程间的通知，但其设计更侧重于事件通知和等待，难以直接实现多读者并发、单作者独占且有优先权的复杂读写锁逻辑。
核心机制：自定义读写锁 RWLock
为了解决上述挑战，我们可以设计一个自定义的读写锁 RWLock。本方案的核心思想是利用 multiprocessing.JoinableQueue 作为读者与作者之间通信和同步的桥梁，实现一种合作式的读写控制。
RWLock 类设计理念
RWLock 的设计遵循以下原则：
立即学习“Python免费学习笔记（深入）”；
- 读者等待作者：读者在开始读取前，会等待作者发出“新数据已准备好”的信号。
- 作者等待读者：作者在开始写入前，会等待所有读者完成对上一轮数据的读取。
- 作者优先权：作者可以通过设置一个“停止”标志，请求正在读取的读者尽快结束当前操作并释放资源，以便作者能快速获得独占权。
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
工作流程解析
-
初始化 (__init__)：
- 为每个预期读者创建一个 JoinableQueue。这些队列初始为空。
- _queue_count 用于确保每个读者进程/线程在首次调用 acquire_for_reading 时能获得一个唯一的队列。
- _stop 是一个共享的 Value，作为作者请求读者立即停止的标志。
-
读者获取锁 (acquire_for_reading)：
- 每个读者首次调用时，会从 _queues 列表中分配一个专属的 JoinableQueue。
- 读者调用 queue.get()，由于队列初始为空，读者进程会在此处阻塞，等待作者放入数据。
-
读者释放锁 (release_for_reading)：
- 读者完成读取后，调用 queue.task_done()。这会通知其专属队列，之前由作者放入的那个任务（即唤醒信号）已经处理完毕。
-
作者获取锁 (acquire_for_writing)：
- 作者遍历所有读者的 JoinableQueue，并调用 queue.join()。
- queue.join() 会阻塞，直到队列中所有通过 put() 添加的任务都被 task_done() 标记为完成。这意味着作者会等待所有读者完成对上一轮数据的读取。
- 如果 immediate 参数为 True，作者会设置 _stop.value = 1，向读者发出立即停止的信号。
-
作者释放锁 (release_for_writing)：
- 作者完成写入后，首先重置 _stop.value = 0。
- 然后，作者向每个读者的 JoinableQueue 中 put(None)。这会解除之前 acquire_for_reading 中读者 queue.get() 的阻塞状态，允许读者开始读取新数据。
-
作者优先权 (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()
注意事项与最佳实践
- 读者协作性：作者的“立即”优先权依赖于读者进程/线程的协作。如果读者在执行长时间任务时不检查 is_stop_posted() 标志，作者将无法强制其立即释放资源，只能等待读者自然完成。因此，在实际应用中，读者任务应设计成可中断的，并在关键点检查此标志。
- 数据一致性：此 RWLock 确保了写入时的独占性，即当作者写入时，没有读者正在读取，也没有其他作者（本设计只支持一个作者）正在写入。同时，它也保证了读者总是读取到作者完整写入后的数据，避免了读取到不一致的中间状态。
- 资源管理：务必确保 acquire_for_reading() 和 acquire_for_writing() 之后，都对应地调用 release_for_reading() 和 release_for_writing()，即使在异常情况下也应如此（例如使用 try…finally 结构），以避免死锁或资源泄露。 4