数据采集​​和处理架构

Question

我正在磨练我的 Python 技能，并开始学习将 websockets 作为一种教育工具。 因此，我正在处理通过 websocket 每毫秒接收到的实时数据。我想以一种干净而全面的方式分离它的采集/处理/绘图。采集和处理至关重要，而绘图可以每约 100 毫秒更新一次。

A) 我假设原始数据以恒定速率到达，每毫秒。

B)如果处理速度不够快 (>1ms)，则跳过忙时到达的数据并与 A)保持同步

C) 每隔约 100 毫秒左右，获取最后处理的数据并绘制它。

我猜最小的工作示例会像这样开始：

import threading

class ReceiveData(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def receive(self):
        pass


class ProcessData(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def process(self):
        pass


class PlotData(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def plot(self):
        pass

从这个开始（这甚至是正确的方法吗？），我如何将原始数据从ReceiveData 传递给ProcessData，并定期传递给PlotData？如何保持执行同步，并每毫秒或每 100 毫秒重复调用一次？

谢谢。

Answer 1

我认为您使用线程接收和处理数据的一般方法很好。对于线程之间的通信，我建议采用生产者-消费者方法。 Here is a complete example 使用 Queue 作为数据结构。

在您的情况下，您希望跳过未处理的数据并仅使用最新的元素。为了实现这一点，collections.deque（请参阅 documentation）可能是您更好的选择 - 另请参阅 this discussion。

d = collections.deque(maxlen=1)

生产者端会像这样将数据附加到双端队列：

d.append(item)

消费者端的主循环可能如下所示：

while True:
    try:
        item = d.pop()
        print('Getting item' + str(item))
    except IndexError:
        print('Deque is empty')
    # time.sleep(s) if you want to poll the latest data every s seconds

您可以将ReceiveData 和ProcessData 功能合并到一个类/线程中，并在此类和PlotData 之间仅使用一个双端队列。