如何让Python中维护记录的字典实现线程安全？含读写锁及MVCC疑问

一、读写锁实现：支持多线程并发读 #

Python标准库没有原生读写锁，但可通过基础锁+计数器手动实现，核心逻辑是允许多个读操作并发执行，写操作独占资源，适合读多写少的场景。

代码实现 #

import threading

class Records:
    def __init__(self):
        self.map = {}  # ID -> list of [year, location]
        self.read_lock = threading.Lock()
        self.write_lock = threading.Lock()
        self.read_count = 0

    def _acquire_read(self):
        # 第一个读操作抢占写锁，后续读仅累加计数器
        with self.read_lock:
            self.read_count += 1
            if self.read_count == 1:
                self.write_lock.acquire()

    def _release_read(self):
        # 最后一个读操作释放写锁
        with self.read_lock:
            self.read_count -= 1
            if self.read_count == 0:
                self.write_lock.release()

    def record(self, ID, year, location):
        # 写操作独占写锁，阻塞所有读写请求
        with self.write_lock:
            if ID not in self.map:
                self.map[ID] = []
            self.map[ID].append([year, location])

    def find_history(self, ID, year):
        # 读操作获取读锁，允许多线程并发执行
        self._acquire_read()
        try:
            if ID not in self.map:
                return []
            results = []
            # 遍历期间写操作被阻塞，无需担心列表被修改
            for record in self.map[ID]:
                if record[0] <= year:
                    results.append(record.copy())  # 返回副本避免外部修改内部数据
            return results
        finally:
            self._release_read()

工作原理 #

• 多个读线程可同时持有读锁，只要无写操作执行；
• 写操作需等待所有读操作完成后才能执行，执行期间会阻塞所有新的读写请求；
• 读操作结束后逐步释放资源，最后一个读线程释放写锁，让等待的写操作执行。

二、MVCC（多版本并发控制）实现：无锁读写分离 #

MVCC核心是写操作不修改原数据，而是生成新版本；读操作读取当前最新版本快照，实现读写完全不阻塞，适合读写都频繁的场景。

代码实现 #

import threading

class Records:
    def __init__(self):
        self._map = {}  # 存储当前最新版本的映射
        self._write_lock = threading.Lock()  # 仅保证版本替换的原子性

    def record(self, ID, year, location):
        while True:
            # 获取当前版本的映射
            current_map = self._map
            # 复制目标ID的记录列表，避免修改原数据
            current_list = current_map.get(ID, [])
            new_list = current_list.copy()
            new_list.append([year, location])
            # 生成新的映射版本
            new_map = current_map.copy()
            new_map[ID] = new_list

            # 原子性替换旧版本：若期间无其他线程修改_map，则替换成功
            with self._write_lock:
                if self._map is current_map:
                    self._map = new_map
                    break
            # 若_map已被其他线程修改，重试整个流程

    def find_history(self, ID, year):
        # 读操作直接读取当前版本快照，无需加锁
        current_map = self._map
        current_list = current_map.get(ID, [])
        results = []
        for record in current_list:
            if record[0] <= year:
                results.append(record.copy())
        return results

工作原理 #

• 每次写操作生成全新的列表和映射版本，原数据不会被修改；
• 读操作读取的是某个时刻的快照，完全不受写操作影响；
• 仅用一个锁保证版本替换的原子性，避免多个写操作同时覆盖版本。

方案对比 #

内容的提问来源于stack exchange，提问作者Jone Shi