本文深入探讨了在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
