Python列表原子操作与多线程：是否需用threading.Lock保障安全？

结论 #

是的，两个线程都需要使用threading.Lock，才能确保列表更新对遍历线程的原子性，避免遍历过程中出现异常、数据不一致或逻辑错误。

原因分析 #

1. 单个操作原子性≠复合操作原子性
   Python列表的部分单个操作（比如append()、pop()、索引赋值）是原子的，但遍历（比如for item in lst）、批量增删这类复合操作由多个步骤组成，不属于原子操作。当遍历线程正在遍历列表时，修改线程的增删操作会直接打乱遍历逻辑：

   • 遍历列表时，Python会生成基于索引的迭代器，每次迭代取当前索引元素并递增索引。若此时修改线程删除元素，可能导致遍历跳过元素；若添加元素，可能引发迭代器重复访问或索引越界。
   • 极端场景下，还可能触发RuntimeError（比如迭代器检测到列表大小异常变化）。

2. 原子性要求“不可分割”
   要实现你需要的原子性，意味着遍历线程要么看到完全未更新的列表，要么看到完全更新后的列表，绝对不能看到增删操作执行到一半的中间状态。这就要求：

   • 修改线程执行增删操作时必须持有锁，保证整个更新过程不被打断；
   • 遍历线程在整个遍历周期内也必须持有锁，保证遍历期间列表不会被修改。

示例代码 #

import threading

lst = []
lock = threading.Lock()

# 遍历线程函数
def traverse_list():
    while True:
        with lock:  # 遍历全程持有锁
            for pattern in lst:
                # 应用正则表达式到对象的逻辑
                print(f"应用正则: {pattern}")
        # 适当休眠，避免过度占用锁
        threading.sleep(1)

# 更新线程函数
def update_list():
    count = 0
    while True:
        with lock:  # 更新操作全程持有锁
            lst.append(f"pattern_{count}")
            if count > 5:
                lst.pop(0)
        count += 1
        threading.sleep(2)

# 启动线程
t1 = threading.Thread(target=traverse_list)
t2 = threading.Thread(target=update_list)
t1.start()
t2.start()

优化建议 #

• 用with lock自动管理锁的获取与释放，比手动调用acquire()/release()更安全，能避免异常导致的锁泄漏。
• 尽量缩短锁的持有时间：如果正则匹配操作耗时较长，可先在锁内复制列表内容，释放锁后再处理复制的副本，减少线程阻塞：

def traverse_list():
    while True:
        current_patterns = []
        with lock:
            current_patterns = lst.copy()  # 锁内快速复制，立即释放锁
        for pattern in current_patterns:
            # 执行耗时的正则匹配操作
            print(f"应用正则: {pattern}")
        threading.sleep(1)

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