面试题：Python的Flask与Django文件上传性能优化及分布式场景处理

高并发场景下Flask和Django文件上传功能性能优化

优化文件读取：

Flask：在Flask中，request.files对象默认将整个文件读入内存。对于大文件，可以使用stream模式，将文件直接写入磁盘或存储系统，避免内存占用。例如：

from flask import Flask, request
app = Flask(__name__)

@app.route('/upload', methods=['POST'])
def upload():
    file = request.files['file']
    file.save('/path/to/save/' + file.filename)
    return 'File uploaded successfully'

Django：Django的UploadedFile对象也有类似情况。可以使用temporary_file_path()方法获取临时文件路径，然后进行后续处理，减少内存开销。

from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt

@csrf_exempt
def upload(request):
    if request.method == 'POST':
        file = request.FILES['file']
        with open('/path/to/save/' + file.name, 'wb+') as destination:
            for chunk in file.chunks():
                destination.write(chunk)
        return HttpResponse('File uploaded successfully')

使用异步处理：

Flask：可以结合asyncio和aiofiles库实现异步文件写入。首先安装aiofiles，然后使用如下代码：

import asyncio
import aiofiles
from flask import Flask, request

app = Flask(__name__)

@app.route('/upload', methods=['POST'])
async def upload():
    file = request.files['file']
    async with aiofiles.open('/path/to/save/' + file.filename, 'wb') as f:
        await f.write(await file.read())
    return 'File uploaded successfully'

Django：Django 3.1+支持异步视图。可以使用async/await语法和aiofiles实现异步文件处理。首先在urls.py中确保正确配置异步视图，然后在视图函数中：

import asyncio
import aiofiles
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt

@csrf_exempt
async def upload(request):
    if request.method == 'POST':
        file = request.FILES['file']
        async with aiofiles.open('/path/to/save/' + file.name, 'wb') as f:
            for chunk in file.chunks():
                await f.write(chunk)
        return HttpResponse('File uploaded successfully')

缓存和压缩：

Flask：可以使用Flask - Caching库进行缓存，减少重复文件上传的处理。例如，对于已经上传过的相同文件，可以直接从缓存中获取。同时，可以使用gzip压缩减少网络传输大小。

from flask import Flask, request
from flask_caching import Cache

cache = Cache(app, config={'CACHE_TYPE':'simple'})

@app.route('/upload', methods=['POST'])
@cache.cached(query_string=True)
def upload():
    file = request.files['file']
    file.save('/path/to/save/' + file.filename)
    return 'File uploaded successfully'

Django：Django有内置的缓存机制。可以在settings.py中配置缓存，然后在视图中使用cache_page装饰器。对于压缩，可以启用django - compress - css和django - compress - js等库对静态文件进行压缩。

from django.views.decorators.cache import cache_page
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt

@csrf_exempt
@cache_page(60 * 15)  # 缓存15分钟
def upload(request):
    if request.method == 'POST':
        file = request.FILES['file']
        with open('/path/to/save/' + file.name, 'wb+') as destination:
            for chunk in file.chunks():
                destination.write(chunk)
        return HttpResponse('File uploaded successfully')

适应分布式存储（以MinIO为例）

Flask与MinIO集成：

安装minio库：pip install minio。
修改上传代码：

from flask import Flask, request
from minio import Minio
from minio.error import S3Error

app = Flask(__name__)
client = Minio(
   "minio.example.com",
   access_key="ACCESS_KEY",
   secret_key="SECRET_KEY",
   secure=True
)

@app.route('/upload', methods=['POST'])
def upload():
    file = request.files['file']
    try:
        client.put_object(
            "your - bucket - name",
            file.filename,
            file.stream,
            file.content_length,
            file.content_type
        )
        return 'File uploaded to MinIO successfully'
    except S3Error as err:
        return f"Error: {err}"

Django与MinIO集成：

安装minio库：pip install minio。
创建一个自定义的文件存储类，在storages.py中：

from django.core.files.storage import Storage
from minio import Minio
from minio.error import S3Error

class MinIOStorage(Storage):
    def __init__(self):
        self.client = Minio(
           "minio.example.com",
           access_key="ACCESS_KEY",
           secret_key="SECRET_KEY",
           secure=True
        )

    def _save(self, name, content):
        try:
            self.client.put_object(
                "your - bucket - name",
                name,
                content,
                content.size,
                content.content_type
            )
            return name
        except S3Error as err:
            raise Exception(f"Error: {err}")

    def url(self, name):
        try:
            return self.client.presigned_get_object(
                "your - bucket - name",
                name
            )
        except S3Error as err:
            raise Exception(f"Error: {err}")

在settings.py中配置使用自定义存储：

DEFAULT_FILE_STORAGE = 'your_app_name.storages.MinIOStorage'

修改视图函数：

from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt

@csrf_exempt
def upload(request):
    if request.method == 'POST':
        file = request.FILES['file']
        from your_app_name.storages import MinIOStorage
        storage = MinIOStorage()
        storage._save(file.name, file)
        return HttpResponse('File uploaded to MinIO successfully')

星途面试题库

面试题：Python的Flask与Django文件上传性能优化及分布式场景处理

知识考点

面试题答案

高并发场景下Flask和Django文件上传功能性能优化

适应分布式存储（以MinIO为例）