Python读写锁实现与优化解析
本文深入解析Python中读写锁的实现机制。针对标准库未提供读写锁的情况，从基础实现出发，分析其公平性缺陷，并进一步探讨读优先与写优先策略。通过引入条件变量和状态计数，重构出支持自定义优先级的读写锁，解决读锁长期占用问题，提升并发性能。
Python读写锁基础概念
起步
Python内置的多线程模型中并未直接提供读写锁。相较于单纯的互斥锁，读写锁具有更高的适用性：它允许多个线程同时以读模式持有锁，但仅允许一个线程以写模式持有锁。
通俗而言，当不存在写锁时，任意线程均可获取读锁；而写锁具有排他性，必须等待所有读锁释放后，才能由单一线程获取。
基础读写锁实现分析
简单的实现
import threading
class RWlock(object):
def __init__(self):
self._lock = threading.Lock()
self._extra = threading.Lock()
self.read_num = 0
def read_acquire(self):
with self._extra:
self.read_num += 1
if self.read_num == 1:
self._lock.acquire()
def read_release(self):
with self._extra:
self.read_num -= 1
if self.read_num == 0:
self._lock.release()
def write_acquire(self):
self._lock.acquire()
def write_release(self):
self._lock.release()上述代码展示了一种基础的读写锁实现。其中，self.read_num 用于记录获取读锁的线程数量。由于该属性属于临界区资源，对其修改需加锁保护，因此代码中引入了额外的锁 self._extra 来保护内部数据状态。
然而，该锁存在公平性问题。理想状态下，无论线程执行读操作还是写操作，获取锁的机会应均等。但观察代码可知，读请求会立即执行 self.read_num += 1，无论是否真正获得锁；而写请求必须等待 read_num 降为 0 才能尝试获取锁。
这导致只要存在读请求或锁未被占用，写权限便难以获得。因此，我们需要采取措施避免读模式锁被长期占用，以保障写操作的及时性。
读写锁优先级策略
读写锁的优先级
读写锁通常分为“读优先”和“写优先”两种策略。前文所述的简单实现属于典型的读优先模式。
若需改为写优先，可改为记录写线程的引用计数。当读写同时竞争时，让写线程增加写计数，使得读线程因需判断写引用计数（若不为0则等待）而难以获取读锁。这部分代码此处不再赘述。
然而，这种硬编码策略缺乏灵活性。我们无需维护两个相似的读写锁类，而是希望重构代码，使其具备更强的通用性和扩展性。
支持自定义优先级的优化实现
改进
为了实现支持自定义优先级的读写锁，我们需要记录等待中的读写线程数量，并引入两个 threading.Condition 对象来处理优先级的通知逻辑。此时，引用计数的语义得以扩展：正数表示正在执行读操作的线程数，负数（最多为-1）表示正在执行写操作的线程数。
在获取读锁时，首先检查是否存在等待的写线程。若无，则直接执行读操作；若有，则等待读计数加 1 后，通过 Condition 通知继续；当读计数减 1 且引用计数加 1 后，若条件不满足则循环等待。
在获取写锁时，若锁未被占用，引用计数减 1；若已被占用，则等待写线程数加 1，并通过写条件 Condition 等待通知。
释放读模式和写锁的逻辑一致，均需根据当前状态判断并通知对应的 Condition 对象。
class RWLock(object):
def __init__(self):
self.lock = threading.Lock()
self.rcond = threading.Condition(self.lock)
self.wcond = threading.Condition(self.lock)
self.read_waiter = 0 # 等待获取读锁的线程数
self.write_waiter = 0 # 等待获取写锁的线程数
self.state = 0 # 正数：表示正在读操作的线程数 负数：表示正在写操作的线程数（最多-1）
self.owners = [] # 正在操作的线程id集合
self.write_first = True # 默认写优先，False表示读优先
def write_acquire(self, blocking=True):
# 获取写锁只有当
me = threading.get_ident()
with self.lock:
while not self._write_acquire(me):
if not blocking:
return False
self.write_waiter += 1
self.wcond.wait()
self.write_waiter -= 1
return True
def _write_acquire(self, me):
# 获取写锁只有当锁没人占用，或者当前线程已经占用
if self.state == 0 or (self.state < 0 and me in self.owners):
self.state -= 1
self.owners.append(me)
return True
if self.state > 0 and me in self.owners:
raise RuntimeError('cannot recursively wrlock a rdlocked lock')
return False
def read_acquire(self, blocking=True):
me = threading.get_ident()
with self.lock:
while not self._read_acquire(me):
if not blocking:
return False
self.read_waiter += 1
self.rcond.wait()
self.read_waiter -= 1
return True
def _read_acquire(self, me):
if self.state < 0:
# 如果锁被写锁占用
return False
if not self.write_waiter:
ok = True
else:
ok = me in self.owners
if ok or not self.write_first:
self.state += 1
self.owners.append(me)
return True
return False
def unlock(self):
me = threading.get_ident()
with self.lock:
try:
self.owners.remove(me)
except ValueError:
raise RuntimeError('cannot release un-acquired lock')
if self.state > 0:
self.state -= 1
else:
self.state += 1
if not self.state:
if self.write_waiter and self.write_first: # 如果有写操作在等待（默认写优先）
self.wcond.notify()
elif self.read_waiter:
self.rcond.notify_all()
elif self.write_waiter:
self.wcond.notify()
read_release = unlock
write_release = unlock以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
