使用concurrent.futures.ThreadPoolExecutor的线程安全问题咨询

代码片段 #

import concurrent.futures
import time
from random import random

class Test(object):
    def __init__(self):
        self.my_set = set()

    def worker(self, name):
        temp_set = set()

        temp_set.add(name)
        temp_set.add(name*10)
        time.sleep(random() * 5)
        temp_set.add(name*10 + 1)

        self.my_set = self.my_set.union(temp_set) # question 1
        return name

    def start(self):
        result = []
        names = [1,2,3,4,5,6,7]
        with concurrent.futures.ThreadPoolExecutor(max_workers=len(names)) as executor:
            futures = [executor.submit(self.worker, x) for x in names]
            for future in concurrent.futures.as_completed(futures):
                result.append(future.result()) # question 2

技术疑问 #

1. 标记为“question 1”的代码行是否可能导致self.my_set损坏？我认为union操作是原子性的，但赋值操作是否存在问题？
2. 标记为“question 2”的代码行是否存在问题？我认为列表的append操作是原子性的，所以或许该操作是安全的？

解答 #

问题1：self.my_set = self.my_set.union(temp_set)的线程安全问题 #

会出现数据丢失，但不会导致集合结构损坏（比如变成不可用状态）。具体原因：

• union操作本身是原子的，它仅读取当前self.my_set的状态，生成一个全新集合，不会修改原集合。
• 但赋值self.my_set = ...不是原子操作。假设有两个线程同时执行这段代码：
  1. 线程A读取当前self.my_set为状态S，执行union得到新集合S_A
  2. 线程B同时读取self.my_set也为状态S，执行union得到新集合S_B
  3. 线程A先完成赋值，self.my_set变为S_A
  4. 线程B后完成赋值，直接将self.my_set覆盖为S_B，线程A的修改完全丢失

解决办法是给self.my_set的操作加锁，比如使用threading.Lock，让读取、生成新集合、赋值的整个流程变成原子操作，避免并发覆盖。

问题2：result.append(future.result())的线程安全问题 #

你的判断是对的，这个操作是安全的。在CPython中，list.append()对应单个字节码指令，而全局解释器锁（GIL）会保证单个字节码指令不会被线程切换打断，所以多线程调用append不会出现数据混乱、元素丢失或重复的情况。

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