Python Django应用如何常驻单个daemon线程运行后台任务

你不需要用递归实现循环，用threading模块配合Django的AppConfig钩子就能实现全局唯一的守护线程，资源占用极低。

核心实现逻辑 #

• 用全局状态标记避免线程重复初始化
• 用while循环+休眠替代递归，不会出现调用栈溢出、内存泄漏问题
• 在Django的AppConfig的ready()方法中完成线程初始化，保证应用启动时自动拉起线程

代码实现 #

首先在对应应用的apps.py文件中编写如下代码：

import threading
import time
import requests
from django.apps import AppConfig

# 全局标记：记录守护线程是否已启动
_daemon_thread_started = False
# 停止事件：用于优雅终止线程
_stop_signal = threading.Event()

def your_request_task(target_url):
    # 循环执行任务，收到停止信号时退出
    while not _stop_signal.is_set():
        try:
            requests.get(target_url)
            # 此处添加你的后续处理逻辑
        except Exception as e:
            # 捕获异常避免单次请求错误导致线程直接退出
            print(f"请求执行出错：{str(e)}")
        # 调整此处的休眠时间，单位为秒，根据你的业务需求设置任务执行间隔
        time.sleep(60)

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app' # 此处替换为你自己的应用名

    def ready(self):
        global _daemon_thread_started
        # 线程已启动则直接返回，避免重复创建
        if _daemon_thread_started:
            return
        _daemon_thread_started = True
        # 初始化守护线程，主线程退出时会自动销毁该线程
        task_thread = threading.Thread(
            target=your_request_task,
            args=("https://your-target-url.com",), # 替换为实际请求地址
            daemon=True
        )
        task_thread.start()

注意事项 #

• 开发环境下Django的runserver默认会启动两个进程（用于代码修改热重载），会导致线程被初始化两次，启动开发服务时添加--noreload参数即可解决：python manage.py runserver --noreload
• 生产环境如果使用多Worker部署（如Gunicorn设置多个Worker），每个Worker进程都会独立启动一个守护线程，如果需要整个服务全局唯一的执行实例，需要额外引入Redis、数据库等实现分布式锁来控制执行权限。

内容的提问来源于stack exchange，提问作者Богдан Божић