Python多进程/多线程读写锁实现：高效并发读与独占写-Python教程-PHP中文网

Python多进程/多线程读写锁实现：高效并发读与独占写

本文深入探讨了在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, 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 = 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, Lock
from queue import Queue
from threading import local
import time

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

        self._local_storage = local()
        self._num_readers = num_readers

        # 在多线程中，这些变量不需要是multiprocessing.Value
        self._queue_count = 0 
        self._stop = 0 
        self._lock = Lock() 
        # 使用threading.Queue
        self._queues = [Queue(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]
                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 = 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('按回车键终止程序:\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多进程/多线程读写锁实现：高效并发读与独占写的详细内容，更多请关注php中文网其它相关文章！