背景

由于celery最小只能到分钟级别，但是日常有需要每隔10秒同步一遍获取远程系统能力

实现

python3
""" Auto reload celery runner script
Usage: python manage.py sync_perms
See: https://stackoverflow.com/a/49166246/2544762
"""
import signal
import traceback
from datetime import datetime
from logging import getLogger
from time import sleep
from typing import Dict

from django.apps import apps
from django.conf import settings
from django.core.management.base import BaseCommand

from mobile.tasks import job_watch_sonic_devices

logger = getLogger(__name__)


class Command(BaseCommand):
    def handle(self, *args, **options):
        # 退出标记
        flag_terminate = False

        # 退出信号处理（Ctrl + C），使得可以正常结束不用 kill
        def signal_handler(signal, frame):
            nonlocal flag_terminate
            flag_terminate = True

        signal.signal(signal.SIGINT, signal_handler)
        signal.signal(signal.SIGTERM, signal_handler)

        # 死循环轮询定时任务
        while not flag_terminate:
            logger.info(f'定时任务开始执行：{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
            try:
                job_watch_sonic_devices()
            except Exception as e:
                print(f'watch sonic e: {e}')
                traceback.print_exception(e)
            # 整体执行时间需要2秒，这里设置8秒，基本10秒内一个轮回
            sleep(8)

1、信号处理

python3
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

捕获 Ctrl + C (SIGINT) 和 终止信号 (SIGTERM)。

使用 nonlocal flag_terminate 让信号处理函数能修改外部变量，确保可以优雅退出循环，而不是被强制 kill。

优点：

运行中按 Ctrl+C 可以安全结束（会在下个循环退出）。

避免直接 kill -9 导致状态不一致。

python3
flag_terminate = False

def signal_handler(signal, frame):
    nonlocal flag_terminate
    flag_terminate = True

底层流程（信号机制是怎么帮忙的）：

当用户 Ctrl+C（SIGINT）或 kill（SIGTERM）时，不是立刻杀掉进程，而是执行 signal_handler：

把 flag_terminate 设成 True。

主循环每轮都会检查 flag_terminate：

如果 True，退出循环，任务结束前可以优雅清理（比如保存状态、断开连接）。

这样：

正在执行的那一轮任务会先跑完。

然后在下一轮循环检查到退出标志，再干净地退出。

简单理解：

信号处理就是给死循环任务加了一个“安全刹车”，让你按下 Ctrl+C 时，不是直接翻车，而是慢慢停下来。

是否同步？

如果只看 Python 解释器层面，当收到信号时，信号处理函数（signal_handler）的执行是同步的。

解释器会在执行字节码的“安全点”检测是否有挂起的信号：

如果有，就暂停当前的 Python 代码执行；

同步调用你注册的 signal.signal(...) 处理函数；

处理函数执行完后，回到原来的位置继续执行

2、技术要点总结

• 自定义 Django 管理命令：可直接用 python manage.py 启动，不依赖外部调度器。
• 信号安全退出：通过 signal.signal + 外部变量标志实现优雅退出。
• 异常隔离：任务出错不会中断整个进程。
• 简单定时调度：用 sleep() 模拟固定周期调度（适合简单场景，不适合精确到秒的生产任务）。
• 日志追踪：方便监控执行情况。
• 可替代 Celery beat：适合轻量任务或测试环境，但在高并发/分布式场景下不如 Celery 稳定