如何在Django线程中实现执行结果的Flash消息通知？

你遇到的问题核心在于：Django的messages是依赖当前请求的Session上下文工作的。当你的主线程返回redirect后，整个请求周期就结束了——Django会把Session的修改写入存储，然后销毁当前请求的上下文。而后台线程此时再去操作request对象的messages，这些修改根本不会被持久化到用户的Session里，自然刷新页面也看不到。

最适合小项目的方案是用缓存存储任务结果，再通过前端定时轮询获取结果并展示。下面是具体实现步骤：

1. 修改原视图，生成任务ID并缓存结果 #

我们需要给每个线程任务生成唯一ID，把任务结果存在缓存里，同时把ID传给前端页面：

from threading import Thread
from django.contrib import messages
from django.core.cache import cache
import uuid

def long_time_view(request, pk):
    # 生成唯一任务ID，用来关联线程任务和前端查询
    task_id = str(uuid.uuid4())
    # 先标记任务为"进行中"，缓存1小时超时
    cache.set(f"task_{task_id}", "pending", timeout=3600)

    def thread_func(task_id):
        success = False
        try:
            # 这里替换成你的长时间任务逻辑
            # ------------------------------
            # stuff...
            # 模拟任务执行成功/失败
            success = True
            # ------------------------------
        finally:
            # 任务完成后把结果存入缓存，5分钟超时
            if success:
                cache.set(f"task_{task_id}", {"status": "success", "message": "Hurrah."}, timeout=300)
            else:
                cache.set(f"task_{task_id}", {"status": "error", "message": "Upss."}, timeout=300)

    messages.info(request, "Started Work.")
    # 把task_id传给线程
    thread = Thread(target=thread_func, args=(task_id,))
    thread.start()

    # 重定向时把task_id附加到返回URL的参数中
    return_url = request.GET.get("return_to", "/")
    separator = "&" if "?" in return_url else "?"
    return redirect(f"{return_url}{separator}task_id={task_id}")

2. 新增任务状态查询视图 #

写一个简单的接口，让前端可以查询任务结果：

from django.http import JsonResponse
from django.core.cache import cache

def check_task_status(request):
    task_id = request.GET.get("task_id")
    if not task_id:
        return JsonResponse({"status": "invalid", "message": "No task ID provided"})
    
    task_result = cache.get(f"task_{task_id}")
    if task_result == "pending":
        return JsonResponse({"status": "pending"})
    elif task_result:
        # 返回结果后立刻删除缓存，避免重复显示
        cache.delete(f"task_{task_id}")
        return JsonResponse(task_result)
    else:
        return JsonResponse({"status": "expired", "message": "Task result expired"})

记得在urls.py里给这个视图加路由：

path('check-task-status/', views.check_task_status, name='check_task_status'),

3. 前端添加轮询逻辑 #

在返回的页面（比如基础模板里）加入JS代码，自动查询任务状态并展示消息：

<script>
// 从URL参数中获取task_id
function getUrlParam(name) {
    const urlParams = new URLSearchParams(window.location.search);
    return urlParams.get(name);
}

const taskId = getUrlParam('task_id');
if (taskId) {
    // 每2秒查询一次任务状态
    const pollInterval = setInterval(() => {
        fetch(`/check-task-status/?task_id=${taskId}`)
            .then(response => response.json())
            .then(data => {
                if (data.status === 'success') {
                    showMessage(data.message, 'success');
                    clearInterval(pollInterval);
                } else if (data.status === 'error') {
                    showMessage(data.message, 'error');
                    clearInterval(pollInterval);
                } else if (data.status === 'expired') {
                    showMessage('Task result expired', 'warning');
                    clearInterval(pollInterval);
                }
                // 状态为pending时继续轮询
            })
            .catch(error => {
                console.error('Error checking task:', error);
                clearInterval(pollInterval);
            });
    }, 2000);
}

// 模拟Django messages的样式显示消息（假设用Bootstrap）
function showMessage(text, type) {
    const messageDiv = document.createElement('div');
    messageDiv.className = `alert alert-${type} alert-dismissible fade show`;
    messageDiv.innerHTML = `
        ${text}
        <button type="button" class="btn-close" data-bs-dismiss="alert"></button>
    `;
    // 把消息加到页面顶部（比如Django通常放messages的位置）
    const messageContainer = document.querySelector('.messages') || document.body;
    messageContainer.insertBefore(messageDiv, messageContainer.firstChild);
    // 5秒后自动关闭
    setTimeout(() => {
        messageDiv.remove();
    }, 5000);
}
</script>

如果你的项目需要更实时的体验，可以用Django Channels实现WebSocket推送——线程任务完成后直接主动把消息推送给前端，不需要轮询。不过这个方案需要配置ASGI服务器，相对复杂一些，适合中大型项目。

内容的提问来源于stack exchange，提问作者Eduardo Alejandro Leyva