没有阻塞或不受控制的队列增长的 Python 线程安全访问？答案

【问题标题】：Python thread-safe access without blocking or uncontrolled queue growth?没有阻塞或不受控制的队列增长的 Python 线程安全访问？
【发布时间】：2014-11-11 16:22:03
【问题描述】：

我有一个硬件传感器。我需要连续读取传感器并对前 3 秒的读数进行一些数学运算。（我读得越快，样本越多，结果就越好，所以我想不断地调用我的函数。）我需要在请求时向用户（通过网络调用）提供当前结果。结果可能会在亚秒级的基础上发生显着变化。用户可以尽可能快地或非常不频繁地请求更新。

我目前有以下代码，它可以工作。

from threading import Thread
from MySensorReadAndMath import MySensorReadAndMath

class ManageSensor(Thread):
    def __init__(self):
        Thread.__init__(self)
        self.result={}
        self.mySensorReadAndMath = MySensorReadAndMath()
    def updateResult(self):
        result = {}
        result['value1'] = self.mySensorReadAndMath.value1
        result['value2'] = self.mySensorReadAndMath.value2
        return result
    def getResult(self):
        return self.result
    def run(self):
        while True:
            self.mySensorReadAndMath.update()
            self.result = self.updateResult()

在单独的一段代码中，我创建了类，使用 Flask 创建对 ManageSensor.getResult() 的 Web 调用以返回结果，然后启动线程。

这是线程安全的吗？（我可以得到当前的 value1 和之前的 value2，还是损坏的 self.result？）如果它不是线程安全的，怎么会我让它线程安全而不阻塞？

我考虑的具体事项：

首先将值复制到 updateResults() 中的本地 result 是否有任何好处（与直接将所有内容直接放入 self.result in updateResult()
如果我在更新 self.result 时加了锁，那么我的 run() 循环或我的 getResult() 可能会阻止我这样做不想。
如果我将结果推送到 run() 中的队列并在 getResult() 中弹出，我将遇到队列大小问题，因为我永远不会调用getResult() 与 self.mySensorReadAndMath.update() 运行一样快。
我是不是想多了？ :)

【问题讨论】：

标签： python multithreading

【解决方案1】：

通常我会尝试使用内置的 Queue 类，但这种情况可能太复杂了。

如果我理解正确，无论用户提出多少请求，您都需要准确存储三秒的数据。您可以利用 Python 总是通过引用访问对象的事实来解决这个问题。分配引用是 Python 中的原子操作，因此您可以安全地使用单例来存储最新结果。有两个条件：

结果发布后不得对成员进行分配
结果的用户必须在请求期间检索一次且仅一次：如果再次检索结果，则结果可能已更改。

所以使用单例来强制执行此行为：

class Buffer:
    ''' A simple buffer that stores exactly one value '''
    latest = None
    @staticmethod
    def onNewReading(*args, **kwds): # or any parameters one fancies
        # Pack the results in a single object and store it
        Result.latest = (args, kwds)
    @staticmethod
    def onUserRequest():
        return Result.latest

写一个值如下：

def process():
    # Do the measurement
    # Do the calculation
    Buffer.onNewReading(data1, data2, etc)

使用值：

def handleRequest():
    results = Buffer.onUserRequest()
    # Format a response, using the results
    if not results:
        return 'No data available'  # or any other useful error message
    return response(results[0][0], results[0][1])

【讨论】：

【解决方案2】：

这不是线程安全的。

客户端线程可以在 updateResult 运行时调用 getResult，因此结果可能为空、半满或包含旧值。

创建另一个对象并将其分配给 result 将限制半填充数据的可能性，因此这是一个加号。

队列解决方案还需要某种锁定

【讨论】：