Python async generator无时间漂移、不计调用次数的定间隔实现咨询

以下是满足需求的修改后代码，完全无需计数变量、无时间漂移，总运行时长稳定接近2.0s：

import asyncio
import time
rx_data = list()
EOT = 20

async def serial_sequence():
    '''async generator function that simulates incoming sequence of serial data'''
    interval = 0.1
    start_time = time.time()
    next_run = start_time + interval

    while True:
        now = time.time()
        dt = now - (next_run - interval)
        delay = max(0, next_run - now)
        print('dt: {}'.format(dt))
        print('delay2: {}'.format(delay))
        await asyncio.sleep(delay) #simulated IO delay
        current_val = round((next_run - start_time) / interval)
        next_run += interval
        yield current_val

async def read_serial1():
    gen = serial_sequence()
    while(True):
        data = await gen.__anext__()
        rx_data.append(data)
        print('read_serial1:', data)
        if data == EOT:
            break
    return rx_data

async def main():
    start = time.time()
    task1 = asyncio.create_task(read_serial1())
    await(task1)
    stop = time.time()
    print('Elapsed: {}'.format(stop-start))

if __name__ == '__main__':
    asyncio.run(main())

关键改动说明 #

• 完全移除了计数变量num：通过初始启动时间和每次触发的锚点时间的差值，自动计算当前返回的序列值，不需要统计生成器的调用次数
• 从根源避免时间漂移：维护next_run作为下一次触发的绝对时间锚点，每次触发后直接累加固定间隔0.1s，不管单次任务处理、事件循环调度耗时多久，永远对齐初始的时间步长，不会产生累计误差
• 增加异常兼容：如果事件循环被其他任务占用过久，导致当前时间已经超过预期触发时间，会自动将sleep时长设置为0，避免参数报错同时快速补触发消除滞后

内容的提问来源于stack exchange，提问作者mr_js