如何实现快速增长列表的线程安全读取？

一、核心前提说明 #

普通Python列表无法直接在多进程间共享——每个进程拥有独立内存空间，跨进程读写普通列表会导致数据不一致。必须使用multiprocessing.Manager提供的进程安全容器，或基于共享内存的结构，才能实现可靠的跨进程数据交互。

二、基于列表的优化方案 #

如果偏好使用列表结构，可通过以下方式避免收集操作阻塞数据追加：

1. 索引式增量收集 #

通过进程安全的索引变量记录上次收集的位置，仅复制新增元素，大幅缩短锁持有时间：

from multiprocessing import Manager, Process, Value
import time

def recorder(shared_list):
    while True:
        shared_list.append(f"item {time.time()}")
        time.sleep(0.001)  # 模拟快速追加

def collector(shared_list, last_idx):
    while True:
        time.sleep(1)  # 模拟定期收集
        with last_idx.get_lock():
            current_len = len(shared_list)
            # 仅复制从上一次收集到当前的新元素
            collected = shared_list[last_idx.value:current_len]
            last_idx.value = current_len
        print(f"Collected {len(collected)} new items")
        # 此处处理collected内的元素

if __name__ == "__main__":
    with Manager() as manager:
        shared_list = manager.list()
        # 用进程安全的整数记录上次收集的末尾索引
        last_idx = Value('i', 0)
        p1 = Process(target=recorder, args=(shared_list,))
        p2 = Process(target=collector, args=(shared_list, last_idx))
        p1.start()
        p2.start()
        p1.join()
        p2.join()

锁仅在获取当前长度和更新索引时短暂持有，几乎不会影响记录进程的追加操作。

2. 双缓冲列表（最优解） #

通过原子替换列表的方式，让收集操作完全不阻塞追加：记录进程始终往"活跃列表"写数据，收集时将活跃列表替换为新空列表，单独处理旧列表。

from multiprocessing import Manager, Process
import time

def recorder(shared_dict):
    while True:
        shared_dict['active'].append(f"item {time.time()}")
        time.sleep(0.001)

def collector(shared_dict):
    while True:
        time.sleep(1)
        # 原子替换活跃列表，操作耗时极短
        old_list = shared_dict['active']
        shared_dict['active'] = manager.list()
        # 处理旧列表时，记录进程已在新列表追加，完全无阻塞
        print(f"Collected {len(old_list)} items")
        # 此处遍历old_list做后续处理

if __name__ == "__main__":
    with Manager() as manager:
        shared_dict = manager.dict()
        shared_dict['active'] = manager.list()
        p1 = Process(target=recorder, args=(shared_dict,))
        p2 = Process(target=collector, args=(shared_dict,))
        p1.start()
        p2.start()
        p1.join()
        p2.join()

这种方式彻底消除了收集操作对追加的影响，也无需担心列表扩容的内存重组问题——内存重组在Manager的服务器进程内部完成，不影响客户端进程的读写。

三、替代数据结构推荐 #

如果不想基于列表实现，可选择更适配场景的专用容器：

• multiprocessing.Queue：适合FIFO场景，记录进程用put追加元素，收集进程可循环get批量取出（注意empty()方法并非绝对可靠，需结合异常处理）。
• Manager.deque：双端队列，append和popleft均为原子操作，适合从头部取元素的场景，批量收集时可结合索引或双缓冲优化锁时间。

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