如何使用Python创建自定义定时器触发的IoT Edge函数，实现无漂移定时采集

原代码时间漂移原因 #

• 仅在程序启动时计算了一次目标时间，后续固定睡眠300秒，没有考虑GatherData的执行耗时、系统调度的延迟，误差会不断累计
• asyncio.gather(asyncio.sleep(...), GatherData())的写法会让睡眠和采集同时执行，采集耗时会直接挤占下一次的等待时间，加速漂移

修正后实现代码 #

from azure.iot.device.aio import IoTHubModuleClient
import asyncio
from datetime import datetime, timedelta

async def GatherData():
    # 此处保留你原有采集逻辑
    pass

async def main():
    sample_interval = 5  # 采集间隔，单位：分钟
    # 初始化IoT Hub客户端，补充原代码缺失的初始化逻辑
    module_client = IoTHubModuleClient.create_from_edge_environment()
    
    try:
        while True:
            # 检查客户端连接状态
            print("module is connected: " + str(module_client.connected))
            if not module_client.connected:
                await module_client.connect()
            
            # 计算下一个5分钟整的UTC时间
            now = datetime.utcnow()
            remainder = now.minute % sample_interval
            if remainder == 0 and now.second == 0:
                next_run = now
            else:
                # 对齐到下一个间隔的0秒0微秒位置
                add_minutes = sample_interval - remainder
                next_run = now.replace(second=0, microsecond=0) + timedelta(minutes=add_minutes)
            
            # 动态计算需要等待的时长
            wait_seconds = (next_run - datetime.utcnow()).total_seconds()
            if wait_seconds > 0:
                await asyncio.sleep(wait_seconds)
            
            # 到点后执行采集任务
            await GatherData()

    except Exception as e:
        print("Unexpected error %s " % e)
        raise
    finally:
        await module_client.shutdown()

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

实现逻辑说明 #

• 每次循环重新计算下一次执行的准点时间，强制对齐到5分钟倍数的0秒位置，不受上一次执行耗时、系统调度延迟的影响，不会出现累计漂移
• 动态计算每次的等待时长，哪怕某次采集耗时几十秒，下一次仍然会自动对齐到下一个5分钟整点
• 如果设备开启了NTP时间同步，系统时间被校准后也会自动适配新的时间，不需要额外调整逻辑

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