Python3.7+下如何为嵌套字典的指定条目加锁？

方案1：锁字典与数据结构一一映射 #

最直接的方式是维护一个和原数据结构完全对应的锁字典，每个需要保护的列表对应一把threading.Lock（若需支持同一线程重复加锁，可改用threading.RLock）。

示例代码 #

import threading

# 原始嵌套字典数据
data = {
    "key1": {
        "key1_level1": [],
        "key1_level2": [],
    }
}

# 对应结构的锁字典，每个列表配一把锁
locks = {
    "key1": {
        "key1_level1": threading.Lock(),
        "key1_level2": threading.Lock(),
    }
}

# 写入操作示例
def add_to_list(top_key, sub_key, item):
    with locks[top_key][sub_key]:
        data[top_key][sub_key].append(item)

# 读取操作示例
def get_list_length(top_key, sub_key):
    with locks[top_key][sub_key]:
        return len(data[top_key][sub_key])

优势 #

• 逻辑直观，完全实现细粒度锁定，操作某一列表时不会阻塞其他列表的读写
• 借助with语句自动管理锁的获取与释放，避免手动解锁遗漏导致的死锁问题

方案2：封装带锁的列表类 #

把列表和对应的锁封装成自定义类，让数据与锁绑定，调用方无需手动管理锁，更易维护。

示例代码 #

import threading

class LockedList:
    def __init__(self):
        self._inner_list = []
        self._lock = threading.Lock()
    
    # 实现需要的列表操作，统一加锁
    def append(self, item):
        with self._lock:
            self._inner_list.append(item)
    
    def __len__(self):
        with self._lock:
            return len(self._inner_list)
    
    def pop(self, index=-1):
        with self._lock:
            return self._inner_list.pop(index)

# 初始化时直接用LockedList替代普通列表
data = {
    "key1": {
        "key1_level1": LockedList(),
        "key1_level2": LockedList(),
    }
}

# 使用时直接调用方法，无需关心锁的存在
data["key1"]["key1_level1"].append("test_item")
print(len(data["key1"]["key1_level1"]))

优势 #

• 封装性强，锁逻辑被隐藏在类内部，降低调用方出错概率
• 可按需扩展列表操作方法，统一加锁逻辑，避免重复代码

注意事项 #

• 若需动态向字典中添加新列表，方案2更便捷，直接创建LockedList实例即可；方案1需同步在锁字典中添加对应锁
• 确保所有操作目标列表的代码都正确获取对应锁，遗漏锁会引发线程安全问题

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