Python多进程/多线程读写锁实现：高效管理共享资源与写者优先策略-Python教程-PHP中文网

Python多进程/多线程读写锁实现：高效管理共享资源与写者优先策略

本文探讨了在python多进程/多线程环境中，如何实现一个高效的读写锁机制，以支持一个写入者和多个读取者对共享资源的并发访问。核心挑战在于允许并发读取的同时，确保写入操作的独占性和高优先级。文章提出了一种自定义的`rwlock`解决方案，通过结合`multiprocessing.joinablequeue`（或`queue.queue`）和共享标志，实现了读者之间的并发性、写者的独占性以及在需要时写者能够请求读者尽快释放资源的策略。

在多进程或多线程编程中，当多个执行单元需要访问同一个共享资源时，同步机制是必不可少的。一个常见的场景是“一写多读”模式：一个写入者负责更新数据，而多个读取者则并发地读取数据。理想情况下，我们希望：

并发读取： 多个读取者可以同时访问共享资源，以提高效率。
独占写入： 写入者在修改数据时必须拥有独占访问权限，以防止数据不一致。
写入者优先： 当写入者需要写入时，应能尽快获得资源，甚至中断正在进行的读取操作。

Python标准库中的multiprocessing.Condition或threading.Condition虽然可以用于复杂的同步，但直接实现上述“一写多读”且带有写入者优先级的模式会比较复杂，尤其是在需要中断读者操作的场景下。例如，简单的Condition通常需要读者在完成读取后通知写入者，或者写入者通知读者有新数据，但无法直接强制读者在读取过程中立即停止并释放资源。

读写锁（RWLock）的设计与实现

为了满足上述需求，我们可以设计一个自定义的读写锁（Read-Write Lock）类。本方案的核心思想是为每个读者分配一个独立的同步队列，并引入一个共享标志来允许写入者请求读者立即停止。

核心组件

我们的RWLock类将包含以下关键组件：

立即学习“Python免费学习笔记（深入）”；

_num_readers: 预期的读者数量。
_local_storage: threading.local对象，用于为每个读者进程/线程存储其专属的队列。
_queues: 一个列表，包含num_readers个multiprocessing.JoinableQueue（或queue.Queue），每个读者一个。
_queue_count: 一个计数器，用于分配队列给读者。
_stop: 一个共享的标志（multiprocessing.Value或普通int），写入者通过设置此标志来请求读者停止。
_lock: 一个互斥锁（multiprocessing.Lock或threading.Lock），用于保护_queue_count的并发访问。

工作机制

1. 读者获取读取权限 (acquire_for_reading)

每个读者首次调用时，会被分配一个专属的JoinableQueue。
读者会对其专属队列执行阻塞的get()操作。这意味着读者会一直等待，直到写入者向其队列中放入一个项目。
这种机制确保了读者在读取前等待写入者完成当前写入并发布新数据。

2. 读者释放读取权限 (release_for_reading)

读者完成数据处理后，调用task_done()方法。这会通知队列，一个先前放入的任务已完成。
task_done()与join()配合使用，允许写入者等待所有读者完成其任务。

3. 写入者获取写入权限 (acquire_for_writing)

写入者在尝试获取独占写入权限时，会遍历所有读者的队列，并对每个队列调用join()方法。
join()会阻塞，直到队列中的所有项目都被get()并调用了task_done()。这意味着写入者会等待所有当前正在读取的读者完成其任务。
如果immediate参数为True，写入者会设置_stop标志为1。这会向读者发出信号，请求它们尽快停止当前读取操作。

4. 写入者释放写入权限 (release_for_writing)

火龙果写作

用火龙果，轻松写作，通过校对、改写、扩展等功能实现高质量内容生产。

106

查看详情

写入者完成写入后，会重置_stop标志为0。
然后，写入者会遍历所有读者的队列，并向每个队列中放入一个None（或其他任意值）。这会解除所有等待中的读者的get()阻塞，允许它们开始读取新数据。

5. 读者检查停止信号 (is_stop_posted)

读者在进行长时间的读取操作时，应周期性地调用is_stop_posted()来检查写入者是否请求立即停止。
如果_stop标志被设置为1，读者应尽快中断当前读取并释放权限。这是一个协作式的机制，读者需要主动配合。

多进程环境下的RWLock实现

在多进程环境中，我们需要使用multiprocessing模块提供的同步原语，如multiprocessing.Process、multiprocessing.Lock、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 必须是正整数。')
        self._local_storage = local() # 用于存储每个进程/线程的专属队列
        self._num_readers = num_readers
        self._queue_count = Value('i', 0) # 共享的队列分配计数器
        self._stop = Value('i', 0)       # 共享的停止标志
        self._lock = Lock()              # 保护 _queue_count 的锁
        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):
        """
        获取独占写入权限。
        如果 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自带的锁

def reader_task(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_task(rw_lock, shared_data):
    while True:
        # 当 shared_data.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():
    num_readers = 3
    rw_lock = RWLock(num_readers)
    shared_data = SharedData()

    # 创建读者进程
    for id in range(1, num_readers + 1):
        Process(target=reader_task, args=(rw_lock, id, shared_data), daemon=True).start()

    # 创建写入者进程
    Process(target=writer_task, args=(rw_lock, shared_data), daemon=True).start()

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

if __name__ == '__main__':
    main()

登录后复制

上述代码示例展示了RWLock在多进程环境中的应用。读者进程通过acquire_for_reading等待新数据，并在处理数据时通过is_stop_posted检查是否需要中断。写入者进程通过acquire_for_writing等待所有读者完成，或在特定条件下（如shared_data.value.value == 3）请求读者立即停止。

运行输出示例：

按 Enter 键终止:
写入者写入: 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
写入者写入: 4 在 1704820192.1564832  # 注意此处写入时间间隔变短，因为读者被中断
读者 1 收到停止请求，中断读取。
读者 1 完成处理数据: 4
读者 2 收到停止请求，中断读取。
读者 2 完成处理数据: 4
读者 3 收到停止请求，中断读取。
读者 3 完成处理数据: 4
写入者写入: 5 在 1704820195.2668517
...

登录后复制

从输出可以看出，当shared_data.value达到3时，写入者会设置immediate=True，导致读者在完成当前循环前检查到停止信号并中断读取，从而使得写入者能够更快地获取写入权限，完成第4次写入。

多线程环境下的RWLock实现

如果仅在多线程环境中使用，可以替换multiprocessing模块的特定类为threading模块的对应类，并简化共享变量的定义。

from threading import Thread, Lock, local
from queue import Queue # 线程安全的队列
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
        self._queue_count = 0 # 普通整数，线程内共享
        self._stop = 0        # 普通整数，线程内共享
        self._lock = Lock()   # 线程锁
        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

### 示例用法 ###

# 共享数据类，普通类成员在线程间共享
class SharedValue:
    def __init__(self):
        self.value = 0

def reader_thread_task(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_task(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_threading():
    num_readers = 3
    rw_lock = RWLockMultiThreading(num_readers)
    shared_data = SharedValue()

    for id in range(1, num_readers + 1):
        Thread(target=reader_thread_task, args=(rw_lock, id, shared_data), daemon=True).start()
    Thread(target=writer_thread_task, args=(rw_lock, shared_data), daemon=True).start()

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

if __name__ == '__main__':
    main_threading()

登录后复制

注意事项与总结

协作式中断： is_stop_posted()机制依赖于读者进程/线程的协作。如果读者不定期检查此标志，或者在检查后未能及时响应中断请求，写入者将无法实现“立即”获取独占权限。对于关键的实时性要求，需要确保读者任务能够快速响应中断。
性能考量： 写入者在acquire_for_writing中必须等待所有读者完成当前的数据处理（通过queue.join()）。这意味着写入者的速度会受到最慢的读者任务的影响。在设计读者任务时，应尽量使其处理时间可控。
资源管理： 确保共享数据在进程/线程之间正确传递和访问。在多进程中，需要使用multiprocessing.Value、multiprocessing.Array或共享内存等机制。在多线程中，普通对象成员即可在线程间共享。
死锁防范： 读写锁的设计应避免死锁。本方案中，读者只在get()时阻塞，写入者只在join()时阻塞，且写入者释放时会唤醒所有读者，结构相对简单，不易产生死锁。
适用场景： 这种自定义的读写锁特别适用于需要高并发读取，但写入操作较少且需要独占和优先级的场景。它提供了一种灵活的方式来平衡读写操作的性能与数据一致性。

通过上述RWLock的实现，我们能够有效地管理共享资源的并发访问，在允许并发读取的同时，确保写入操作的独占性和高优先级，从而构建出更健壮、高效的多进程/多线程应用程序。

以上就是Python多进程/多线程读写锁实现：高效管理共享资源与写者优先策略的详细内容，更多请关注php中文网其它相关文章！