Python异步队列同时使用get/put

Question

我想运行一个 Python 脚本，它可以模拟真实世界的时间序列数据并实时处理数据。也就是说，从给定的数据集(大约 8 小时长的测量)中，我想每秒获取一秒长的数据，并在读取每一秒的数据后立即处理。在现实世界中，数据将使用一些检测器每秒收集一次。对于这个任务，我决定使用 Python asyncio模块。下面基本上是我想出的。

import scipy
import numpy as np
import asyncio
import time
import queue

q = queue.Queue()

async def get_data (data):

    while True:
        await asyncio.sleep(1)
        q.put(data[idx,:])
        idx += 1
        #Each row of data is read every second. 

async def run_algorithm ():

    while True:

        if q.empty() == True:
            await asyncio.sleep(1)

        data_read = q.get(block = False)

        #I do something here with the read data


async def main (data):

    feed_data = asyncio.create_task(get_data(data))
    process_data = asyncio.create_task(run_algorithm ())
    await asyncio.gather(feed_data, process_data)

虽然它似乎工作正常，但问题是#I 在这里对读取数据部分做了一些事情。这就是我使用我的算法实时处理数据的地方。当我使用一种快速的算法时，它运行得很好。它应该每秒读取一次数据，当队列中没有数据时，run_algorithm 函数会等待一秒钟。

但是，当我有一个慢速算法时，它不会每秒读取数据。如果算法需要 0.5 秒运行，则在 1.5 秒内读取下一个数据，而不是 1 秒。

就像 get_data 函数随着 run_algorithm 部分变慢而变慢一样。无论 run_algorithm 部分需要多长时间，有没有办法让 get_data 每秒读取一次数据？

Answer 1

如评论中所示，您需要使用异步队列，在这种情况下，您不需要 run_algorithm 中的 sleep 。 :

q = asyncio.Queue()

async def get_data (data):
    while True:
        await asyncio.sleep(1)
        await q.put(data[idx,:])
        idx += 1
        #Each row of data is read every second. 

async def run_algorithm ():
    while True:
        data_read = await q.get()
        #I do something here with the read data

However, when I have a slow algorithm, it does not read the data every second

听起来您的算法包含受 CPU 限制或以其他方式阻塞的代码。由于 asyncio 是单线程的，这会导致事件循环停止。要修复它，您应该在单独的线程中运行阻塞代码。 Asyncio 包含一个用于此目的的实用程序， run_in_executor .例如，使用 time.sleep(1) 模拟阻塞/CPU 绑定(bind)调用(故意不使用 asyncio.sleep 来模拟阻塞)，这是从 asyncio 调用它的方式:

        #time.sleep(1)   # this is forbidden
        # instead, invoke blocking code like this:
        loop = asyncio.get_event_loop()
        await loop.run_in_executor(None, time.sleep, 1)