asyncio中如何安全实现共享列表的元素移除操作

问题本质 #

asyncio 是单线程协作式并发模型，协程只会在 await 位置主动让出事件循环控制权。如果遍历共享列表时碰到await点让出控制权，其他协程刚好对列表做增删操作修改了列表结构，就会直接抛出RuntimeError: list changed size during iteration异常。

可选实现方案 #

根据业务对一致性、性能的要求，可以选以下三种常用方案： #

方案1：遍历前拷贝列表副本（实现成本最低） #

遍历操作前先对原列表做一次浅拷贝，后续遍历完全基于副本执行，原列表的任何增删都不会影响当前遍历流程。
示例代码：

a = [] # List[Connection]

async def foo():
    # list(a) 生成原列表的浅拷贝，仅拷贝列表内的元素引用，开销极低
    for v in list(a):
        await v.send('msg')

async def bar():
    await SomeAsyncFunc()
    # 直接修改原列表不会影响已经开始的遍历操作
    a.pop(0)

适用场景：遍历过程不需要感知列表新增的元素，对数据一致性要求不高，想要最低改造成本。 #

方案2：用 asyncio.Lock 做临界区互斥（一致性最强） #

用asyncio提供的协程级锁，把所有涉及列表结构修改、遍历的操作都纳入锁保护，保证同一时间只有一个协程操作列表结构。
注意不要把长耗时的await操作放在锁持有阶段，避免锁粒度过长阻塞其他协程，可以先在锁内拷贝出要操作的对象，释放锁后再执行耗时逻辑：

import asyncio
a = [] # List[Connection]
list_op_lock = asyncio.Lock()

async def foo():
    # 仅在拷贝列表的极短时间内持有锁
    async with list_op_lock:
        send_targets = list(a)
    # 释放锁后再执行耗时的消息发送，不阻塞其他协程增删列表
    for v in send_targets:
        await v.send('msg')

async def bar():
    await SomeAsyncFunc()
    # 修改列表结构时必须持有锁
    async with list_op_lock:
        a.pop(0)

适用场景：对列表数据一致性要求高，需要保证遍历/增删操作的原子性。注意不要使用多线程的threading.Lock，会阻塞整个事件循环。 #

方案3：标记删除替代即时删除（高并发场景性能最优） #

不直接在协程中修改列表结构，所有删除操作先给对应元素打删除标记，遍历的时候自动跳过已标记删除的元素，等遍历结束或者低峰期再统一清理列表里的无效元素。
示例代码：

a = [] # List[Connection]

async def foo():
    for v in a:
        # 跳过已标记为关闭的连接
        if getattr(v, "is_closed", False):
            continue
        await v.send('msg')
    # 遍历完成后一次性清理所有已标记删除的元素
    a[:] = [v for v in a if not getattr(v, "is_closed", False)]

async def bar():
    await SomeAsyncFunc()
    # 不修改列表结构，仅给待删除元素打标记
    a[0].is_closed = True

适用场景：列表增删操作非常频繁，能接受删除操作有短暂延迟，想要最高的并发性能。 #

注意事项 #

• 列表浅拷贝仅复制列表内存储的对象引用，不会深度拷贝对象本身，万级元素以内的拷贝开销可以忽略
• 所有涉及列表长度变化的操作（append/pop/insert/切片赋值等）都属于结构修改，需要和遍历操作做隔离
• 不要在持有锁的阶段执行无超时的网络IO、长耗时计算，会拖慢整个事件循环的调度效率

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