使用匿名threading.Lock()是否始终为错误用法？附代码咨询

你的判断完全正确，这段代码里的with threading.Lock()确实是错误用法，完全起不到线程同步的作用。

问题原因 #

线程锁的核心作用是通过共享同一个锁实例，让多个线程在访问临界资源时互斥执行。而这段代码里每次调用get_batch都会创建一个全新的threading.Lock()对象——不同线程拿到的是完全独立的锁，彼此之间没有任何关联，自然无法阻止多个线程同时进入被锁包裹的代码块。

换句话说，这个锁相当于“假锁”，对self.full_queue、self.input_buffers这些实例级共享资源的访问完全没有线程安全保障。

正确做法 #

应该在类的初始化方法（__init__）中创建一个共享的锁实例，让所有线程共用这一把锁：

class YourClass:
    def __init__(self):
        # 其他初始化逻辑
        self.lock = threading.Lock()  # 初始化共享锁
        self.full_queue = ...
        self.input_buffers = ...
    
    def get_batch(self) -> Union[Tuple[List[int], torch.Tensor], Tuple[None, None]]:
        """Return an inference batch"""
        with self.lock:  # 使用共享锁
            indices: List[int] = []
            for _ in range(self.batch_size):
                try:
                    index = self.full_queue.get(timeout=0.05)
                    indices.append(index)
                except:
                    break

            if indices:
                batch = {
                    key: torch.stack([self.input_buffers[key][index] for index in indices])
                    .to(torch.device('cpu'), non_blocking=True)
                    .unsqueeze(0)
                    for key in self.input_buffers
                }
                return indices, batch
            else:
                return None, None

这样所有调用get_batch的线程都会竞争同一把锁，确保临界区代码（队列取数、缓冲区读取）的执行是互斥的，避免线程安全问题。

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