基于Django实现HTTP大文件上传接收及Celery测试方案咨询

是否必须分片处理大文件？ #

不是必须，分场景判断：

• 如果文件小于服务器单次上传限制（比如Nginx的client_max_body_size、Django的DATA_UPLOAD_MAX_MEMORY_SIZE），直接用普通表单上传即可，无需分片。
• 如果文件超过服务器单次请求限制，或者需要支持断点续传、上传进度展示，那就必须做分片处理。

实现场景与对应方案 #

场景1：小文件（≤服务器单次上传限制） #

直接利用Django原生的文件处理能力：

1. 前端用普通表单或FormData对象发起POST请求。
2. Django视图接收并保存文件：

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
import os

@csrf_exempt
def upload_file(request):
    if request.method == 'POST':
        file_obj = request.FILES.get('file')
        if not file_obj:
            return JsonResponse({'status': 'error', 'msg': '未上传文件'})
        
        # 自定义保存路径，也可配置MEDIA_ROOT使用默认存储
        save_dir = 'uploads'
        os.makedirs(save_dir, exist_ok=True)
        save_path = os.path.join(save_dir, file_obj.name)
        
        # 按块写入，避免内存占用过高
        with open(save_path, 'wb+') as destination:
            for chunk in file_obj.chunks():
                destination.write(chunk)
        return JsonResponse({'status': 'success', 'msg': '文件上传完成'})

注意：若文件超过DATA_UPLOAD_MAX_MEMORY_SIZE（默认2.5MB），Django会自动将文件存到临时目录，无需手动处理；同时要确保服务器（如Nginx）的client_max_body_size配置足够大。

场景2：大文件（>服务器单次限制/需断点续传） #

采用分片上传方案，核心流程：

1. 前端逻辑：
   • 将文件切割为固定大小的分片（比如10MB/片）。
   • 按顺序上传每个分片，附带文件唯一标识（如MD5）、分片索引、总片数。
   • 所有分片上传完成后，向后端发送合并请求。
2. Django后端逻辑：
   • 接收分片并按文件标识+分片索引保存到临时目录。
   • 收到合并请求后，校验所有分片是否齐全，合并为完整文件后清理临时分片。
     合并逻辑示例：

import os
from django.http import JsonResponse

def merge_chunks(request):
    if request.method == 'POST':
        file_md5 = request.POST.get('file_md5')
        total_chunks = int(request.POST.get('total_chunks'))
        filename = request.POST.get('filename')
        
        temp_dir = os.path.join('temp_uploads', file_md5)
        save_path = os.path.join('uploads', filename)
        
        # 合并所有分片
        try:
            with open(save_path, 'wb') as target_file:
                for i in range(total_chunks):
                    chunk_path = os.path.join(temp_dir, f'{i}.part')
                    if not os.path.exists(chunk_path):
                        return JsonResponse({'status': 'error', 'msg': '缺失分片'})
                    with open(chunk_path, 'rb') as chunk_file:
                        target_file.write(chunk_file.read())
                    os.remove(chunk_path)
            os.rmdir(temp_dir)
            return JsonResponse({'status': 'success', 'msg': '文件合并完成'})
        except Exception as e:
            return JsonResponse({'status': 'error', 'msg': str(e)})

场景3：异步处理大文件（可选） #

若上传后需要执行解压、转码等耗时操作，可结合Celery异步执行，避免阻塞HTTP请求。

用Celery测试大文件上传功能 #

Celery适合模拟并发上传、验证异步处理流程，测试思路如下：

1. 模拟并发上传请求 #

批量发起HTTP上传请求，测试服务器并发处理能力：

from celery import shared_task
import requests

@shared_task
def simulate_single_upload(file_path, upload_url):
    with open(file_path, 'rb') as f:
        response = requests.post(upload_url, files={'file': f})
    return {'status_code': response.status_code, 'response': response.json()}

# 调用示例：模拟10个并发上传任务
for _ in range(10):
    simulate_single_upload.delay('/path/to/large_file.bin', 'http://your-django-domain/upload')

2. 测试异步文件处理流程 #

验证上传后的异步处理逻辑（如MD5校验、格式转换）：

@shared_task
def process_uploaded_file(file_path):
    # 模拟文件校验：计算MD5
    import hashlib
    md5_hash = hashlib.md5()
    with open(file_path, 'rb') as f:
        for chunk in iter(lambda: f.read(4096), b''):
            md5_hash.update(chunk)
    return {'file_path': file_path, 'md5': md5_hash.hexdigest()}

# 在上传视图中触发异步任务
def upload_file(request):
    # ...文件保存逻辑...
    process_uploaded_file.delay(save_path)
    return JsonResponse({'status': 'success', 'msg': '文件上传中，后台处理中'})

# 测试时直接调用任务验证结果
task_result = process_uploaded_file.delay('/path/to/uploaded_file')
print(task_result.get())

3. 测试分片上传完整流程 #

模拟分片上传+合并的全流程：

from celery import shared_task
import requests
import hashlib
import os

@shared_task
def simulate_chunked_upload(file_path, chunk_upload_url, merge_url):
    chunk_size = 10 * 1024 * 1024  # 10MB分片
    file_md5 = hashlib.md5(open(file_path, 'rb').read()).hexdigest()
    filename = os.path.basename(file_path)
    total_chunks = (os.path.getsize(file_path) + chunk_size - 1) // chunk_size
    
    # 上传所有分片
    with open(file_path, 'rb') as f:
        for i in range(total_chunks):
            chunk_data = f.read(chunk_size)
            requests.post(
                chunk_upload_url,
                files={'chunk': (f'{i}.part', chunk_data)},
                data={'file_md5': file_md5, 'chunk_index': i, 'total_chunks': total_chunks}
            )
    
    # 发起合并请求
    merge_response = requests.post(
        merge_url,
        data={'file_md5': file_md5, 'total_chunks': total_chunks, 'filename': filename}
    )
    return {'merge_status': merge_response.status_code, 'merge_msg': merge_response.json()}

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