Django使用Celery处理视频转GIF如何避免504超时并返回文件

长耗时的视频下载、转码逻辑放在同步HTTP请求链路里必然会触发网关超时，Celery作为独立运行的异步任务组件，本身和Web服务的请求/响应周期完全隔离，不可能直接向发起请求的用户返回HTTP响应或文件，你现有逻辑只完成了「异步提交任务」这一步，缺失了任务状态跟踪、结果存储、结果交付三个核心环节。
不要靠调大Nginx、Gunicorn超时阈值解决问题，这种方案会占满Web服务连接池，流量稍高就会导致服务不可用。

1. 新增任务持久化模型 #

不要只依赖Celery自带的结果存储，用Django模型关联任务全链路信息，方便做权限校验、进度跟踪、文件管理，核心字段参考：

• celery_task_id：CharField，存Celery返回的任务ID，加唯一索引
• session_key/user：关联提交任务的用户，匿名用户存会话ID，登录用户关联用户表，用于权限校验
• target_url：URLField，存用户提交的待下载视频地址
• status：CharField，枚举值：排队中/下载中/转换中/成功/失败
• gif_file：FileField，存最终生成的GIF文件
• error_msg：TextField，存任务失败时的错误信息
• created_at/finished_at：时间字段，记录任务创建、完成时间

2. 改造提交任务的视图逻辑 #

提交任务后不要直接跳回首页，先创建任务记录、绑定Celery任务ID，再跳转到专属的任务状态页：

def post(self, request):
    form = URLform(request.POST)
    ctx = {'form':form}
    if not form.is_valid():
        return render(request, 'home.html', ctx)
    
    video_url = form.cleaned_data['url']
    # 创建任务记录
    task_obj = GifConvertTask.objects.create(
        target_url=video_url,
        status="pending",
        session_key=request.session.session_key
    )
    # 提交异步任务，传入任务记录ID
    celery_task = download_video.delay(task_obj.id, video_url)
    # 绑定Celery任务ID
    task_obj.celery_task_id = celery_task.id
    task_obj.save()
    
    # 跳转到任务状态页
    return redirect('gif_to_mp4:task_status', task_id=celery_task.id)

注意：你现有代码里的任务名downloand_video存在拼写错误，建议修正为download_video，避免后续排查问题踩坑。

3. 改造Celery任务逻辑 #

任务执行全流程同步更新数据库里的任务状态，最终生成的GIF文件存入Django媒体目录，不要存在临时目录：

import os
from celery import shared_task
from django.core.files import File
from .models import GifConvertTask
# 提前封装好的下载、转码工具函数
from .utils import download_third_party_video, convert_video_to_gif

@shared_task(bind=True)
def download_video(self, task_obj_id, video_url):
    task_obj = GifConvertTask.objects.get(id=task_obj_id)
    try:
        # 更新状态：下载中
        task_obj.status = "downloading"
        task_obj.save()
        temp_video_path = download_third_party_video(video_url)

        # 更新状态：转换中
        task_obj.status = "converting"
        task_obj.save()
        temp_gif_path = convert_video_to_gif(temp_video_path)

        # 保存生成的GIF到模型字段
        with open(temp_gif_path, 'rb') as f:
            task_obj.gif_file.save(f"output_{task_obj.id}.gif", File(f))
        
        # 清理临时文件
        os.remove(temp_video_path)
        os.remove(temp_gif_path)

        # 标记任务成功
        task_obj.status = "success"
        task_obj.save()
    except Exception as e:
        # 标记任务失败，记录错误信息
        task_obj.status = "failed"
        task_obj.error_msg = str(e)
        task_obj.save()
        raise e

4. 实现任务状态查询与前端轮询 #

写一个轻量的状态接口，前端在任务状态页每隔2-3秒轮询一次任务状态，不需要保持长连接，从根本上避免504超时：

后端状态接口示例 #

from django.http import JsonResponse
from .models import GifConvertTask

def task_status(request, task_id):
    # 权限校验：只允许提交任务的会话/用户查看结果
    task_obj = GifConvertTask.objects.get(
        celery_task_id=task_id,
        session_key=request.session.session_key
    )
    # AJAX请求返回JSON状态
    if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
        resp = {
            'status': task_obj.status,
            'error_msg': task_obj.error_msg
        }
        if task_obj.status == 'success':
            resp['gif_url'] = task_obj.gif_file.url
        return JsonResponse(resp)
    # 普通请求返回状态展示页
    return render(request, 'task_status.html', {'task_id': task_id})

前端轮询逻辑示例（放到task_status.html里） #

<div id="status-text">已提交任务，排队中...</div>
<div id="result"></div>

<script>
const TASK_ID = "{{ task_id }}";
function pollStatus() {
    fetch(`/task-status/${TASK_ID}/`, {
        headers: {'X-Requested-With': 'XMLHttpRequest'}
    })
    .then(res => res.json())
    .then(data => {
        if (data.status === 'success') {
            document.getElementById('result').innerHTML = `
                <img src="${data.gif_url}" alt="生成的GIF">
                <p><a href="${data.gif_url}" download>点击下载GIF文件</a></p>
            `;
            document.getElementById('status-text').textContent = "处理完成！";
            return;
        }
        if (data.status === 'failed') {
            document.getElementById('status-text').textContent = `处理失败：${data.error_msg}`;
            return;
        }
        // 更新处理中状态
        const statusMap = {
            'pending': '排队中，请稍候...',
            'downloading': '正在下载源视频...',
            'converting': '正在转换为GIF格式...'
        };
        document.getElementById('status-text').textContent = statusMap[data.status];
        // 2秒后再次查询
        setTimeout(pollStatus, 2000);
    })
}
document.addEventListener('DOMContentLoaded', pollStatus);
</script>

• 配置定时任务，定期清理超过7天的历史任务记录和对应GIF文件，避免磁盘空间被占满
• 下载、转码逻辑必须加超时控制，第三方站点无法访问、转码卡住时直接标记任务失败，避免Celery worker被挂死
• 小流量站点用前端轮询足够用，不需要额外引入WebSocket等组件，实现成本最低、稳定性最高
• 静态文件（生成的GIF）尽量用Nginx或者对象存储承载访问压力，不要让Django处理静态文件请求

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