如何异步卸载Django自定义FileField save方法中的视频转码计算？

方案可行性 #

完全可行，异步处理耗时的转码任务是解决前端请求超时的标准方案。通过将转码逻辑从请求响应流程中剥离，前端能立刻收到上传成功的反馈，后端在后台完成转码后再更新数据库状态，完美规避同步处理的阻塞问题。

是否需要自定义视图？ #

不需要重写整个ViewSet，仅需在原有ModelViewSet的perform_create方法中做修改即可。如果需要更复杂的逻辑（比如返回任务ID），也可以自定义create方法，但基础场景下修改perform_create足够。

具体实现思路 #

1. 调整模型结构 #

新增状态字段跟踪转码进度，拆分原始文件和转码后文件的存储字段，因为转码完成前数据库中不会有转码后的文件：

from django.db import models

class Media(models.Model):
    STATUS_CHOICES = (
        ('pending', '待处理'),
        ('processing', '转码中'),
        ('completed', '完成'),
        ('failed', '失败'),
    )
    original_file = models.FileField(upload_to='media/original/')
    converted_file = models.FileField(upload_to='media/converted/', blank=True, null=True)
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending')
    created_at = models.DateTimeField(auto_now_add=True)

2. 引入任务队列（推荐Celery） #

生产环境下最稳定的异步任务方案是Celery，搭配Redis或RabbitMQ作为消息中间件。以下是基础配置和任务实现：

Celery配置（settings.py）

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

转码任务定义（tasks.py）

from celery import shared_task
from django.core.files import File
import os
from pyffmpeg import FFmpeg
from .models import Media

@shared_task(bind=True)
def convert_video_task(self, media_id):
    media = Media.objects.get(id=media_id)
    try:
        # 更新状态为转码中
        media.status = 'processing'
        media.save(update_fields=['status'])
        
        # 执行转码逻辑
        input_path = media.original_file.path
        output_filename = f"{os.path.splitext(os.path.basename(input_path))[0]}.mp4"
        output_path = os.path.join(os.path.dirname(input_path), output_filename)
        
        ff = FFmpeg()
        ff.convert(input_path, output_path)
        
        # 保存转码后的文件到模型
        with open(output_path, 'rb') as f:
            media.converted_file.save(output_filename, File(f), save=False)
        media.status = 'completed'
        media.save()
        
        # 清理临时文件（可选）
        os.remove(output_path)
    except Exception as e:
        # 转码失败更新状态
        media.status = 'failed'
        media.save(update_fields=['status'])
        # 异常重试（可选）
        raise self.retry(exc=e, countdown=5)

3. 修改ViewSet触发异步任务 #

在ModelViewSet的perform_create方法中，保存原始文件后立刻触发转码任务，无需等待转码完成：

# views.py
from rest_framework import viewsets
from .models import Media
from .serializers import MediaSerializer
from .tasks import convert_video_task

class MediaViewSet(viewsets.ModelViewSet):
    queryset = Media.objects.all()
    serializer_class = MediaSerializer
    
    def perform_create(self, serializer):
        # 保存媒体记录，初始状态为待处理
        media = serializer.save(status='pending')
        # 异步触发转码任务
        convert_video_task.delay(media.id)

4. 前端配合轮询状态 #

Vue端上传后，通过轮询接口获取转码状态，完成后展示视频：

async uploadVideo(file) {
  const formData = new FormData();
  formData.append('original_file', file);
  
  try {
    const { data: media } = await this.$axios.post('/api/media/', formData);
    // 每3秒轮询一次状态
    const statusInterval = setInterval(async () => {
      const { data: updatedMedia } = await this.$axios.get(`/api/media/${media.id}/`);
      if (updatedMedia.status === 'completed') {
        clearInterval(statusInterval);
        this.videoUrl = updatedMedia.converted_file;
      } else if (updatedMedia.status === 'failed') {
        clearInterval(statusInterval);
        alert('视频转码失败，请重新上传');
      }
    }, 3000);
  } catch (err) {
    console.error('上传失败:', err);
  }
}

替代方案（轻量场景） #

如果不想引入Celery，也可以用Django的asyncio结合后台线程，但生产环境不推荐（WSGI服务器可能会终止后台线程）。示例如下：

# views.py
import asyncio
import threading
from rest_framework import viewsets
from .models import Media
from .serializers import MediaSerializer

def run_async_task(coro):
    loop = asyncio.new_event_loop()
    threading.Thread(target=loop.run_until_complete, args=(coro,)).start()

async def async_convert_video(media_id):
    # 这里写异步转码逻辑，和Celery任务类似
    media = Media.objects.get(id=media_id)
    media.status = 'processing'
    media.save(update_fields=['status'])
    # ... 转码代码 ...

class MediaViewSet(viewsets.ModelViewSet):
    queryset = Media.objects.all()
    serializer_class = MediaSerializer
    
    def perform_create(self, serializer):
        media = serializer.save(status='pending')
        run_async_task(async_convert_video(media.id))

注意事项 #

• 确保Django有文件存储目录的读写权限，避免转码时出现权限错误。
• 可以增加转码进度字段，在任务中实时更新进度值，前端轮询时显示进度条。
• 生产环境建议使用Celery+Redis的组合，稳定性和可扩展性更好。

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