数据采集和处理体系结构

Question

我正在提高我的Python技能，并开始学习websockets作为一种教育工具。因此，我正在处理通过websocket接收的每毫秒的实时数据。我想以一种干净和全面的方式来区分它的获取/处理/策划。获取和处理至关重要，而绘图可以每100 be更新一次。

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

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

( C)每~100 it左右，获取最后处理的数据并绘制成图。

我想一个最低限度的工作示例应该是这样开始的：

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

我认为您用于接收和处理数据的线程的一般方法是好的。对于线程之间的通信，我建议采用生产者-消费者的方法。下面是一个完整的例子使用Queue作为数据结构。

在这种情况下，您希望跳过未处理的数据，只使用最近的元素。为了实现这一点，collections.deque (参见文档)可能是一个更好的选择--参见这一讨论。

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