Django模板中后台运行函数时实时更新变量的实现方法

我有一个包含按钮的简单模板：

<a class="button" href="{% url 'function' %}">
    <button>
        Settings
    </button>
</a>

同时在views.py中定义了如下函数：

def function(request):
    context = {}
    object = Runner()
    object.prepate_sth()
    context['tasks'] = runner.tasks.remaining
    try:
        thread = threading.Thread(target=runner.run_sth, args=())
        thread.start()
    except Exception:
        raise Exception
    return render(request, 'home.html', context)

我通过创建独立线程避免阻塞函数执行，后台任务会改变runner.tasks.remaining列表的元素数量，希望在模板中实时显示该变量的变化，请问如何实现？

方法一：AJAX定时轮询（简单易实现） #

通过前端定时发送请求到后端获取最新任务数，是最快捷的实现方式，无需额外依赖。

1. 后端新增接口
   在views.py中添加专门返回剩余任务数的视图：

from django.http import JsonResponse
from threading import Lock

# 全局锁保证多线程下数据安全
task_lock = Lock()

def get_remaining_tasks(request):
    # 确保runner是全局可访问的实例（或通过缓存/数据库获取）
    with task_lock:
        remaining_count = len(runner.tasks.remaining)
    return JsonResponse({'remaining': remaining_count})

在urls.py中配置路由：

path('get-remaining-tasks/', views.get_remaining_tasks, name='get_remaining_tasks'),

2. 前端更新模板
   在home.html中添加显示元素和定时请求的JavaScript：

<!-- 用于显示剩余任务数的容器 -->
<div id="remaining-tasks">剩余任务数：{{ tasks|length }}</div>

<script>
// 每2秒发送一次请求更新数据
setInterval(() => {
    fetch("{% url 'get_remaining_tasks' %}")
        .then(res => res.json())
        .then(data => {
            document.getElementById('remaining-tasks').textContent = `剩余任务数：${data.remaining}`;
        })
        .catch(err => console.error('更新任务数失败:', err));
}, 2000);
</script>

方法二：WebSocket实时推送（低延迟） #

如果需要更低延迟的实时更新，可以使用WebSocket结合Django Channels实现主动推送。

1. 安装配置Channels

pip install channels

在settings.py中添加配置：

INSTALLED_APPS = [
    # ...其他应用
    'channels',
]

ASGI_APPLICATION = '你的项目名.asgi.application'

2. 创建WebSocket消费者
   新建consumers.py：

import json
from channels.generic.websocket import WebsocketConsumer
from threading import Lock

task_lock = Lock()

class TaskConsumer(WebsocketConsumer):
    def connect(self):
        self.accept()
        # 连接成功后发送初始任务数
        with task_lock:
            count = len(runner.tasks.remaining)
        self.send(text_data=json.dumps({'remaining': count}))

    def receive(self, text_data):
        # 接收前端请求后返回最新任务数
        with task_lock:
            count = len(runner.tasks.remaining)
        self.send(text_data=json.dumps({'remaining': count}))

新建routing.py配置WebSocket路由：

from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/tasks/$', consumers.TaskConsumer.as_asgi()),
]

修改asgi.py：

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from . import routing

os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": URLRouter(routing.websocket_urlpatterns),
})

3. 前端WebSocket逻辑

<div id="remaining-tasks">剩余任务数：{{ tasks|length }}</div>

<script>
const ws = new WebSocket(`ws://${window.location.host}/ws/tasks/`);

// 接收后端推送的更新
ws.onmessage = (event) => {
    const data = JSON.parse(event.data);
    document.getElementById('remaining-tasks').textContent = `剩余任务数：${data.remaining}`;
};

// 可选：主动定时请求更新
setInterval(() => {
    ws.send('update');
}, 1000);
</script>

关键注意事项 #

• 线程安全：多线程操作runner.tasks.remaining时必须加锁（如threading.Lock()），避免数据竞争导致的异常。
• 实例共享：确保后台线程和视图访问的是同一个Runner实例，可以将其存储在全局变量、缓存（如Redis）或数据库中。
• 资源清理：使用全局实例时要注意定时清理资源，避免长期运行导致内存泄漏。

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