本文深入探讨了在python多进程或多线程环境中，如何高效地管理一个写入者和多个读取者对共享资源的访问。我们提出并详细实现了一个自定义的读写锁（rwlock），该锁通过巧妙结合`joinablequeue`、`lock`和共享变量，确保了读取者可以并发访问数据，而写入者在需要时能够获得独占且优先的写入权限，同时保证数据一致性。文章提供了针对多进程和多线程环境的完整代码示例及详细解释。
引言：多进程/线程共享资源管理的挑战
在构建多进程或多线程系统时，一个常见场景是存在一个负责更新共享数据的写入者（Writer）和多个并发读取该数据的读取者（Reader）。在这种模式下，核心挑战在于：
数据一致性：当写入者修改数据时，必须阻止任何读取者访问，以避免读取到不完整或不一致的数据。并发性：当没有写入操作时，多个读取者应该能够同时访问数据，以最大化系统吞吐量。写入者优先级：当写入者需要写入时，它应该能够尽快获得独占访问权，甚至可以要求正在进行的读取操作尽快中断并释放资源。
Python标准库中的multiprocessing.Condition对象通常用于复杂的同步场景，但其主要用途是实现等待/通知机制，并不直接提供读写锁的语义，即无法天然支持多个并发读取者。简单的Lock虽然可以保证互斥访问，但会导致读取者也必须串行执行，从而降低并发效率。为了解决这些问题，我们需要一个专门的读写锁机制。
自定义读写锁（RWLock）设计原理
我们提出的RWLock类旨在满足上述需求。其核心思想是为每个读取者分配一个独立的JoinableQueue，并利用JoinableQueue的get()、put()和join()方法来实现复杂的同步逻辑。
核心机制：
读取者获取读取权限 (acquire_for_reading)：
立即学习“Python免费学习笔记（深入）”；
每个读取者首次尝试获取权限时，会被分配一个它专属的JoinableQueue。读取者通过在其队列上调用queue.get()来阻塞，等待写入者发出“可以读取”的信号。一旦get()返回，读取者就获得了读取权限。
读取者释放读取权限 (release_for_reading)：
读取者完成数据处理后，调用queue.task_done()通知其队列，表明已完成对当前数据的处理。
写入者获取写入权限 (acquire_for_writing)：
写入者首先遍历所有读取者的队列，对每个队列调用queue.join()。queue.join()会阻塞直到该队列中所有之前由写入者put进去的任务都被task_done()标记完成。这确保了在写入者开始写入之前，所有读取者都已完成了对上一轮数据的读取。如果immediate=True，写入者会设置一个共享的停止标志（_stop），请求读取者尽快停止当前读取操作。
写入者释放写入权限 (release_for_writing)：
小绿鲸英文文献阅读器
英文文献阅读器，专注提高SCI阅读效率
199 查看详情 
写入者完成数据写入后，将共享停止标志重置为0。然后，写入者遍历所有读取者的队列，对每个队列调用queue.put(None)。这会唤醒所有等待在queue.get()上的读取者，通知它们有新数据可供读取。
读取者协作中断 (is_stop_posted)：
读取者在执行耗时操作时，可以周期性地调用is_stop_posted()检查写入者是否设置了停止标志。如果标志被设置，读取者应尽快中断当前读取并释放权限，以响应写入者的紧急写入请求。
多进程环境下的RWLock实现
在多进程环境中，共享变量需要使用multiprocessing模块提供的特定类型，如multiprocessing.Value和multiprocessing.JoinableQueue，以确保数据在不同进程间正确同步。
from multiprocessing import Process, Lock, Value, JoinableQueuefrom threading import local # 用于存储进程局部变量import timeclass RWLock: def __init__(self, num_readers: int): """ 创建一个支持单个写入者和多个读取者的读写锁。 num_readers: 读取者的数量。 """ if num_readers 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 # 阻塞直到写入者put了一个None，表示有新数据可读 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 # 阻塞直到所有读取者都完成了对之前数据的处理 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表示Value内部不使用锁，由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.value}', flush=True) rw_lock.release_for_reading() time.sleep(0.1) # 短暂休眠，避免忙循环def writer(rw_lock, shared_data): while True: # 当shared_data.value.value == 3时，请求立即写入 rw_lock.acquire_for_writing(immediate=(shared_data.value.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('按回车键终止程序:n')if __name__ == '__main__': main()
运行输出示例及解释：
按回车键终止程序:写入数据: 1 在 1704820185.6386113读者 1 完成处理数据: 1读者 2 完成处理数据: 1读者 3 完成处理数据: 1写入数据: 2 在 1704820188.7424514读者 1 完成处理数据: 2读者 2 完成处理数据: 2读者 3 完成处理数据: 2写入数据: 3 在 1704820191.8461268读者 1 完成处理数据: 3读者 2 完成处理数据: 3读者 3 完成处理数据: 3读者 1 收到停止信号，提前中断。读者 2 收到停止信号，提前中断。读者 3 收到停止信号，提前中断。写入数据: 4 在 1704820192.1564832读者 1 完成处理数据: 4读者 2 完成处理数据: 4读者 3 完成处理数据: 4...
从输出可以看出，通常情况下写入者会等待所有读取者完成当前数据的处理（大约3秒，因为reader 3睡眠时间最长）。但当shared_data.value.value达到3时，写入者调用acquire_for_writing(immediate=True)。此时，读取者会收到停止信号并立即中断其当前读取任务，从而让写入者几乎立即获得写入权限，显著缩短了写入等待时间。
多线程环境下的RWLock实现
如果确定只在多线程环境中使用，可以进行一些优化，将multiprocessing模块的特定类型替换为threading模块和标准Python类型，以提高效率。
from threading import Thread, Lockfrom queue import Queuefrom threading import localimport timeclass RWLockMultiThreading: def __init__(self, num_readers: int): """ 创建一个支持单个写入者和多个读取者的读写锁（多线程版本）。 num_readers: 读取者的数量。 """ if num_readers 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): """ 写入者请求独占写入权限。 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 = 0def 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('按回车键终止程序:n')if __name__ == '__main__': main_thread()
注意事项与最佳实践
数据一致性是首要原则：在任何并发编程场景中，确保数据在修改过程中不被其他线程/进程读取到不一致的状态至关重要。本RWLock设计通过queue.join()机制强制写入者等待所有读取者完成当前数据的处理，从而保证了写入时的独占性。合作式中断：is_stop_posted()机制依赖于读取者的合作。如果读取者不定期检查此标志，或者其任务无法中断，那么写入者即使设置了immediate=True，也可能需要等待读取者自然完成。因此，在设计读取任务时，应确保其具备可中断性，并频繁检查停止标志。性能考量：写入者必须等待所有读取者完成任务，这意味着最慢的读取者会决定写入操作的响应时间。JoinableQueue的开销相对较高，但在需要复杂同步逻辑（如等待所有消费者完成任务）时，它是非常有效的工具。对于读取任务非常短的场景，这种复杂的读写锁可能引入不必要的开销，简单的Lock在某些情况下可能表现更好，但这会牺牲并发读取的能力。threading.local的作用：_local_storage = local()是一个关键设计。它确保了每个进程（或线程）都有自己独立的queue引用。当一个新进程（或线程）首次调用acquire_for_reading()时，它会从共享的_queues列表中获取一个未分配的队列，并将其存储在自己的_local_storage.queue中。这样，每个读取者进程/线程都能独立地操作自己的队列，而不会相互干扰。错误处理：RWLock的构造函数对num_readers进行了简单的验证。在实际应用中，应考虑更全面的错误处理，例如当num_readers不匹配实际启动的读取者数量时可能出现的问题。
总结
本文详细介绍了一种在Python多进程和多线程环境中实现高效读写锁的方案。通过自定义RWLock类，结合JoinableQueue的特性，我们成功解决了在保证数据一致性的前提下，实现多个并发读取者和具有优先级的独占写入者之间的同步问题。这种模式特别适用于读操作远多于写操作，且对读取并发性要求较高的场景。理解并正确运用这种读写锁机制，能够显著提升并发系统的性能和健壮性。
以上就是Python多进程/多线程读写锁实现：高效并发读与独占写的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/580693.html
微信扫一扫 
支付宝扫一扫 
