python - 过滤器线程安全吗

Question

我有一个线程正在更新一个名为 l 的列表。我是否正确地说从另一个线程执行以下操作是线程安全的？

filter(lambda x: x[0] == "in", l)

如果它不是线程安全的，那么这是正确的方法吗:

import threading
import time
import Queue

class Logger(threading.Thread):
    def __init__(self, log):
        super(Logger, self).__init__()
        self.log = log
        self.data = []
        self.finished = False
        self.data_lock = threading.Lock()

    def run(self):
        while not self.finished:
            try:
                with self.data_lock: 
                    self.data.append(self.log.get(block=True, timeout=0.1))
            except Queue.Empty:
                pass

    def get_data(self, cond):
        with self.data_lock: 
            d = filter(cond, self.data)      
        return d 

    def stop(self):
        self.finished = True
        self.join()  
        print("Logger stopped")

其中 get_data(self, cond) 方法用于以线程安全的方式检索 self.data 中数据的一小部分。

Answer 1

首先，回答标题中的问题:filter 只是一个函数。因此，它的线程安全性将取决于您使用它的数据结构。

正如评论中已经指出的，列表操作本身在 CPython 中是线程安全的，并受 GIL 保护，但可以说这只是 CPython 的一个实现细节，您不应该真正依赖它。即使您可以依赖它，它们的某些操作的线程安全也可能并不意味着您所说的那种线程安全:

问题在于使用 filter 对序列进行迭代通常不是原子操作。可以在迭代期间更改序列。根据迭代器底层的数据结构，这可能会导致或多或少的奇怪效果。解决此问题的一种方法是迭代使用一个原子操作创建的序列副本。对 tuple、list、string 等标准序列执行此操作的最简单方法是使用切片运算符，如下所示:

filter(lambda x: x[0] == "in", l[:])

除了这对于其他数据类型不一定是线程安全的之外，还有一个问题:它只是一个浅拷贝。由于您的列表元素似乎也类似于列表，因此另一个线程可以并行执行 del l[1000][:] 以清空其中一个内部列表(在您的浅拷贝中指向以及)。这将使您的过滤器表达式失败并出现 IndexError。

话虽如此，使用锁来保护对您的列表的访问并不可耻，我绝对会推荐它。根据您的数据如何更改以及您如何处理返回的数据，在持有锁的同时深度复制元素并返回这些副本甚至可能是明智的。这样你就可以保证一旦返回，过滤条件不会因为返回的元素而突然改变。

写。您的 Logger 代码:我不是 100% 确定您打算如何使用它，以及在一个队列上运行多个线程并 join 它们是否对您至关重要。我觉得奇怪的是你从不使用 Queue.task_done() (假设它的 self.log 是一个 Queue)。此外，您对队列的轮询可能会造成浪费。如果您不需要线程的 join，我建议至少将锁定获取转向:

class Logger(threading.Thread):
    def __init__(self, log):
        super(Logger, self).__init__()
        self.daemon = True
        self.log = log
        self.data = []
        self.data_lock = threading.Lock()

    def run(self):
        while True:
            l = self.log.get()  # thread will sleep here indefinitely
            with self.data_lock: 
                self.data.append(l)
            self.log.task_done()

    def get_data(self, cond):
        with self.data_lock: 
            d = filter(cond, self.data)
            # maybe deepcopy d here
        return d

您仍然可以在外部执行 log.join() 以确保 log 队列的所有元素都已处理。