如何在浏览器中查看Django应用的任务执行进度？

你的核心问题在于Django的同步视图是请求-响应单次模式，一次请求只能返回一次结果，所以用render或redirect只能展示最终或第一步的状态，没法分步更新页面。要实现实时进度展示，需要结合后端异步任务执行+前端主动查询/后端主动推送的方式，下面是两种最常用的落地方案：

方案一：轮询查询（简单易上手，适合入门） #

这是最容易实现的方案，核心思路是：

• 后端开线程后台执行任务，把进度存在缓存/数据库里
• 前端定时发送请求查询进度，更新页面内容

1. 配置缓存（存储进度） #

生产环境建议用Redis，本地测试可以先用Django内置缓存，在settings.py里配置：

# 本地测试用内存缓存（重启服务会清空）
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
        'LOCATION': 'unique-snowflake',
    }
}

# 生产环境替换为Redis
# CACHES = {
#     'default': {
#         'BACKEND': 'django_redis.cache.RedisCache',
#         'LOCATION': 'redis://127.0.0.1:6379/1',
#         'OPTIONS': {
#             'CLIENT_CLASS': 'django_redis.client.DefaultClient',
#         }
#     }
# }

2. 编写后端视图 #

from django.shortcuts import render
from django.http import JsonResponse
from django.core.cache import cache
import threading
import uuid
import time

# 模拟你的三个任务函数
def function1():
    time.sleep(2)  # 替换为实际耗时操作
    return "End of function1, start of function 2"

def function2():
    time.sleep(2)
    return "End of function2, start of function 3"

def function3():
    time.sleep(2)
    return "End of function3, task completed"

# 后台执行任务的函数
def run_task(task_id):
    # 初始化进度
    cache.set(f"task_{task_id}_progress", "Starting task...", timeout=300)
    cache.set(f"task_{task_id}_completed", False, timeout=300)
    
    # 分步执行并更新进度
    progress = function1()
    cache.set(f"task_{task_id}_progress", progress, timeout=300)
    
    progress = function2()
    cache.set(f"task_{task_id}_progress", progress, timeout=300)
    
    progress = function3()
    cache.set(f"task_{task_id}_progress", progress, timeout=300)
    
    # 标记任务完成
    cache.set(f"task_{task_id}_completed", True, timeout=300)

# 启动任务的视图
def start_task(request):
    if request.method == "POST":
        task_id = str(uuid.uuid4())  # 生成唯一任务ID
        # 开线程后台执行任务，不阻塞响应
        threading.Thread(target=run_task, args=(task_id,)).start()
        return JsonResponse({"task_id": task_id})
    return render(request, "task_progress.html")

# 查询进度的API视图
def get_progress(request, task_id):
    progress = cache.get(f"task_{task_id}_progress", "Waiting for update...")
    completed = cache.get(f"task_{task_id}_completed", False)
    return JsonResponse({"progress": progress, "completed": completed})

3. 配置URL路由 #

在urls.py中添加：

from django.urls import path
from . import views

urlpatterns = [
    path("start-task/", views.start_task, name="start_task"),
    path("progress/<str:task_id>/", views.get_progress, name="get_progress"),
]

4. 前端模板（task_progress.html） #

<!DOCTYPE html>
<html>
<head>
    <title>Task Progress</title>
</head>
<body>
    <button id="startBtn">Start Task</button>
    <div id="progressDisplay" style="margin-top: 20px; font-size: 18px;"></div>

    <script>
        const startBtn = document.getElementById('startBtn');
        const progressDisplay = document.getElementById('progressDisplay');

        startBtn.addEventListener('click', () => {
            startBtn.disabled = true;
            startBtn.textContent = "Task Running...";
            progressDisplay.textContent = "";

            // 发送启动任务请求
            fetch("/start-task/", {
                method: "POST",
                headers: {
                    "X-CSRFToken": "{{ csrf_token }}",
                    "Content-Type": "application/x-www-form-urlencoded",
                }
            })
            .then(res => res.json())
            .then(data => {
                // 每500ms查询一次进度
                const interval = setInterval(() => {
                    fetch(`/progress/${data.task_id}/`)
                    .then(res => res.json())
                    .then(progressData => {
                        progressDisplay.textContent = progressData.progress;
                        // 任务完成后停止查询
                        if (progressData.completed) {
                            clearInterval(interval);
                            startBtn.disabled = false;
                            startBtn.textContent = "Start Task";
                        }
                    });
                }, 500);
            });
        });
    </script>
</body>
</html>

方案二：WebSocket实时推送（实时性更强） #

如果需要进度更新无延迟，可以用Django Channels实现WebSocket双向通信，核心是后端任务执行时主动向前端推送进度。不过这个方案需要配置ASGI服务器，适合对实时性要求高的场景：

1. 安装channels和channels_redis
2. 配置ASGI应用，替换原来的WSGI
3. 编写WebSocket消费者，处理前端连接、接收任务ID、推送进度
4. 后台任务执行时通过Channels向前端发送进度消息

方案选择建议 #

• 如果你是Django新手，优先用轮询方案，代码少、易调试，满足基本需求
• 如果任务执行时长较长、对实时性要求高，再考虑WebSocket方案

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