Python:对大容量实时数据流进行在线处理

Question

我在不同的websockets (传感器数据-每月几个TBs )上输入了一些大容量数据，其中

我希望保证所有数据都存储在中，即使在高负载期间也是如此。

因此，我希望将数据分派到我的数据库和实时处理模块(例如GUI、ML预测等)，其方式是：

它缓冲的数据，以防所述模块的处理太慢，这样当负载减少时，这些模块就可以‘追赶’。

我尝试过的内容:带有队列(来自队列模块或线程模块)的 python线程，但是如果它阻塞了，我无法确保数据不会阻塞，如果它的非阻塞(例如asyncio.Queue)，我会得到竞争条件，事情就会崩溃。

所以，也许我应该使用某种回调方法，但我不知道该寻找什么。我希望问题不要太含糊。--如果有人知道我可以尝试什么--最好只使用python，那将对我有很大的帮助，即使这只是一个想法。

Answer 1

，所以也许我应该使用某种回调方法，但我不知道该寻找什么。

看来你需要一个Future。

• concurrent.futures.Future()
• asyncio.Future()

我尝试了什么:具有队列(来自队列模块或线程模块)的python线程，但是如果它是阻塞的，我不能确保数据不阻塞，如果它的非阻塞(例如asyncio.Queue)，我会得到竞争条件，事情就会崩溃。

您可以尝试使用Queue的非阻塞方法。

• Queue.put_nowait()
• Queue.get_nowait()

以及当queue.Empty或queue.Full异常捕获时查询数据库。