数据采集和处理的体系结构

Question

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

A）我假设原始数据以恒定速率每毫秒到达一次

B）如果处理速度不够快（>；1ms），请跳过忙碌时到达的数据，并与A保持同步）

C）每隔约100ms，获取最后处理的数据并绘制它

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

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)

然后，生产者方将向deque追加数据，如下所示：

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之间只使用一个deque