Python中while True正确用法及7*24小时运行场景实现方案

手写while True循环的合理性分析 #

这种写法只适合临时写脚本做短时间测试，完全不适合7*24小时运行的生产系统，核心问题有四个：

• 存在不可避免的时间漂移：time.sleep(0.5)是在任务执行完之后才等待0.5秒，算上拉流、图像处理的耗时，实际两次处理的间隔必然大于0.5秒，跑上几周累计偏差能到几十分钟，根本达不到稳定2次/秒的要求。
• 容错能力为0：只要某一次拉流因为网络抖动、摄像头临时离线、解码失败抛异常，整个循环直接退出，没有任何兜底能力。
• 无资源管控：如果某次循环异常退出没走到资源释放逻辑，跑久了会出现摄像头句柄泄漏、内存不释放的问题，最后直接把系统资源占满导致服务挂死。
• 多路流适配成本极高：要同时拉多台摄像头的话，要么串行执行延迟叠加，要么自己手写多线程管理逻辑，锁、队列、异常处理都要自己实现，很容易出隐蔽bug。

调度器选型建议 #

非常推荐用成熟的调度框架实现这类固定周期的长稳任务，不用自己重复造轮子处理时间校准、异常隔离、任务防堆积这些通用问题，可靠性比手写裸循环高一个量级。
选调度器的时候注意不要选太轻量的玩具实现，要支持三个核心能力：

• 支持固定间隔触发，能自动校准时间，不会因为任务执行耗时产生漂移
• 支持任务级别的异常捕获，单个任务失败不会影响整体服务运行
• 支持任务堆积策略配置，避免任务执行慢的时候无限排队占满内存

可直接复用的实现示例 #

以下是Python场景下适配多摄像头拉流的实现，已经带了异常兜底、资源释放、防任务堆积的逻辑：

import logging
from apscheduler.schedulers.blocking import BlockingScheduler

# 初始化日志，长稳运行必须留日志方便排查问题
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

def release_camera_resource(camera_id: str):
    """自定义的摄像头资源释放逻辑，实际使用时替换成自己的实现"""
    # 比如关闭已经打开的摄像头连接、释放解码缓存等
    pass

def getimage(camera_id: str):
    """自定义的拉流逻辑，实际使用时替换成自己的实现"""
    # 比如通过rtsp拉取对应摄像头的实时帧
    pass

def action(image):
    """自定义的图像处理逻辑，实际使用时替换成自己的实现"""
    # 比如做目标检测、帧存盘、推流等
    pass

def process_single_camera(camera_id: str):
    """单路摄像头的拉流+处理封装，所有异常全部内部捕获"""
    try:
        image = getimage(camera_id)
        action(image)
        logging.debug(f"摄像头{camera_id}帧处理完成")
    except Exception as e:
        logging.error(f"摄像头{camera_id}处理异常: {str(e)}", exc_info=True)
        # 异常时主动释放资源，避免句柄泄漏
        release_camera_resource(camera_id)

if __name__ == "__main__":
    # 初始化调度器，指定时区避免时间计算错误
    scheduler = BlockingScheduler(timezone="Asia/Shanghai")
    # 配置需要拉取的摄像头列表
    camera_list = ["cam_01", "cam_02", "cam_03", "cam_04"]

    for cam_id in camera_list:
        scheduler.add_job(
            process_single_camera,
            trigger="interval",
            seconds=0.5,  # 间隔0.5秒，对应每秒2次的频率
            args=[cam_id],
            coalesce=True,  # 系统卡顿导致错过多次任务时，只执行最后一次，避免任务堆积
            max_instances=1,  # 同一路摄像头同一时间只允许一个处理任务运行，避免帧排队
            misfire_grace_time=10  # 超过10秒没执行的任务直接丢弃，不补跑
        )

    try:
        logging.info("拉流服务启动")
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        # 服务退出时统一释放所有资源
        for cam_id in camera_list:
            release_camera_resource(cam_id)
        logging.info("服务正常退出")

额外注意事项 #

如果单帧处理的平均耗时已经超过0.4秒，建议把拉流和处理逻辑做解耦：调度器只负责按固定频率把帧拉下来放到队列里，单独开工作线程从队列取帧做处理，避免因为处理太慢导致调度器频繁错过任务触发时间。

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