跨线程对集合/列表增删操作的线程安全实现问询

嘿，针对你提到的线程安全进度跟踪和循环退出的问题，我来给你梳理几个实用的方案，结合Python的特性来解决：

普通的list、set这类容器本身不是线程安全的（多进程场景下更是完全无法直接共享），所以我们要么用内置的线程安全容器，要么手动加锁保护共享状态的访问。

1. 用线程/进程安全的容器（最省心的方案） #

Python标准库提供了现成的安全容器，分两种场景：

多线程场景：用queue.Queue #

队列天然适合“取走元素就不再处理”的需求，queue.Queue的所有操作都是线程安全的，而且判断队列是否为空、取元素的逻辑可以无缝结合：

import threading
import queue

def t1(done_set, todo_queue):
    while True:
        try:
            # 非阻塞取元素，队列为空时抛出异常
            item = todo_queue.get(block=False)
            # 这里写t1专属的处理逻辑
            print(f"t1处理了元素：{item}")
            # 标记为已处理
            done_set.add(item)
            todo_queue.task_done()
        except queue.Empty:
            # 队列为空，退出循环
            break

def t2(done_set, todo_queue):
    while True:
        try:
            item = todo_queue.get(block=False)
            # 这里写t2专属的处理逻辑
            print(f"t2处理了元素：{item}")
            done_set.add(item)
            todo_queue.task_done()
        except queue.Empty:
            break

if __name__ == "__main__":
    done_set = set()
    todo_queue = queue.Queue()
    # 初始化待处理元素
    for num in [1,2,3,4,5,6]:
        todo_queue.put(num)
    
    # 启动线程
    thread1 = threading.Thread(target=t1, args=(done_set, todo_queue))
    thread2 = threading.Thread(target=t2, args=(done_set, todo_queue))
    
    thread1.start()
    thread2.start()
    
    # 等待队列所有任务完成
    todo_queue.join()
    thread1.join()
    thread2.join()
    
    print(f"最终已处理元素：{done_set}")

多进程场景：用multiprocessing.Manager的共享容器 #

如果是用multiprocessing.Process，普通容器没法跨进程共享，得用Manager创建代理容器，它内部自带锁保证进程安全：

from multiprocessing import Process, Manager
import queue

def t1(done_set, todo_queue):
    while True:
        try:
            item = todo_queue.get(block=False)
            print(f"t1处理了元素：{item}")
            done_set.add(item)
            todo_queue.task_done()
        except queue.Empty:
            break

def t2(done_set, todo_queue):
    while True:
        try:
            item = todo_queue.get(block=False)
            print(f"t2处理了元素：{item}")
            done_set.add(item)
            todo_queue.task_done()
        except queue.Empty:
            break

if __name__ == "__main__":
    with Manager() as manager:
        done_set = manager.set()
        todo_queue = manager.Queue()
        for num in [1,2,3,4,5,6]:
            todo_queue.put(num)
        
        p1 = Process(target=t1, args=(done_set, todo_queue))
        p2 = Process(target=t2, args=(done_set, todo_queue))
        
        p1.start()
        p2.start()
        
        todo_queue.join()
        p1.join()
        p2.join()
        
        print(f"最终已处理元素：{done_set}")

2. 手动加锁保护普通容器（灵活度更高） #

如果你不想用队列，也可以自己用锁来保护list或set的访问，核心是所有对共享容器的读写操作都要在锁的保护下进行：

import threading

def t1(todo_list, done_set, lock):
    while True:
        # 先加锁检查并取出元素
        with lock:
            if not todo_list:
                break
            # 从末尾取元素效率更高
            item = todo_list.pop()
        # 耗时的处理逻辑放在锁外面，减少锁持有时间
        print(f"t1处理了元素：{item}")
        # 标记已处理也要加锁
        with lock:
            done_set.add(item)

def t2(todo_list, done_set, lock):
    while True:
        with lock:
            if not todo_list:
                break
            item = todo_list.pop()
        print(f"t2处理了元素：{item}")
        with lock:
            done_set.add(item)

if __name__ == "__main__":
    todo_list = [1,2,3,4,5,6]
    done_set = set()
    lock = threading.Lock()
    
    thread1 = threading.Thread(target=t1, args=(todo_list, done_set, lock))
    thread2 = threading.Thread(target=t2, args=(todo_list, done_set, lock))
    
    thread1.start()
    thread2.start()
    
    thread1.join()
    thread2.join()
    
    print(f"最终已处理元素：{done_set}")

3. 循环退出时机的关键细节 #

退出循环的核心是准确判断“没有待处理元素了”，这里要注意：

• 不要在锁外检查容器是否为空：比如线程A刚检查完todo_list不为空，线程B立刻把最后一个元素取走，线程A再去取就会报错。
• 用队列的queue.Empty异常是最安全的方式，因为get(block=False)是原子操作，队列为空时直接抛出异常，不需要额外检查。
• 如果需要提前终止所有线程，可以用threading.Event（多线程）或multiprocessing.Event（多进程），设置事件后，线程循环里同时检查事件和待处理容器。

额外优化建议 #

• 如果t1和t2处理的是不同类型的元素，可以把元素包装成(类型标记, 元素内容)，线程拿到后判断是否属于自己的处理类型，或者直接分开两个队列，减少竞争。
• 尽量避免在锁内执行耗时操作，锁只用来保护共享状态的读写，这样能大幅提高并发效率。

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