IT面试题|麻瓜面试

Python Web 生态丰富，从 Django 的"电池内置"到 FastAPI 的"现代高性能"。理解框架原理、ORM、中间件、部署是 Python 后端工程师的基本功。

Q1：Django、Flask、FastAPI 三大框架的区别和选型？「🟡 中级」

考察点：考察候选人对 Python Web 生态的整体认知，是否了解主流框架的核心差异和适用场景，筛选只会用某一个框架而缺乏全局视野的人。

参考答案：

Python Web 三大主流框架各有其设计哲学和适用场景：

核心定位对比

维度	Django	Flask	FastAPI
设计哲学	电池内置（Batteries Included）	微内核 + 扩展（Micro + Extensions）	现代高性能（Modern & Fast）
架构模式	MTV（Model-Template-View）	无强制模式，自由组织	函数式 / 类视图，基于 Starlette
异步支持	3.1+ 部分支持（异步视图）	需借助 gevent 等	原生异步（ASGI）
类型提示	逐步引入	无	核心驱动（Pydantic）
自动文档	Admin / DRF 需配置	需扩展（flask-restx）	内置 Swagger UI / ReDoc
ORM	内置 Django ORM	无，需自行集成（SQLAlchemy）	无，需自行集成（SQLAlchemy）
学习曲线	陡峭（概念多）	平缓（核心小）	中等（需懂类型提示 + 异步）
生态丰富度	极高（Django 生态）	高（扩展众多）	快速增长中
性能（基准）	中等	中等	高（接近 Go/Node）
适用场景	企业级应用、CMS、Admin 后台	小型项目、API、原型验证	高性能 API、微服务、异步后端

各框架详解

Django

全栈框架，内置 ORM、Admin 后台、Auth 认证、Form 表单、模板引擎、缓存框架等
遵循 MTV 模式，约定大于配置
代表项目：Instagram、Pinterest、Bitbucket、Mozilla
适合团队协作的中大型项目，开发效率极高

Flask

微框架，核心只提供路由、请求响应处理，其余靠扩展
灵活度极高，开发者可以自由选择技术栈
扩展生态丰富：Flask-SQLAlchemy、Flask-Login、Flask-Migrate 等
适合小型项目、API 服务、快速原型、学习研究

FastAPI

基于 Starlette（ASGI 框架）和 Pydantic（数据验证）构建
类型提示驱动：通过 Python 类型注解自动完成参数验证、序列化、文档生成
原生异步支持，性能接近 Node.js 和 Go
自动生成交互式 API 文档（Swagger UI / ReDoc）
适合构建高性能 API、微服务、异步后端服务

选型建议

项目规模大、团队协作、需要 Admin → Django
项目小、灵活度要求高、快速原型 → Flask
API 优先、性能要求高、异步需求 → FastAPI

实际选型时还需考虑：

团队技术栈熟悉度
项目长期维护成本
社区生态和招聘难度
性能和并发需求

追问延伸：

你在项目中是如何做技术选型的？有哪些考量因素？
FastAPI 比 Flask 快多少？性能瓶颈在哪里？
Django 为什么不适合做微服务？有没有办法把 Django 拆成微服务？
三大框架的中间件机制有什么区别？

Q2：Django 的 MTV 架构和 MVC 有什么区别？「🟢 校招/初级」

考察点：考察对 Web 经典架构模式的理解，以及 Django 设计思想的认知，筛选连基本架构都讲不清楚的人。

参考答案：

MVC 模式

MVC（Model-View-Controller）是经典的软件架构模式：

Model（模型）：负责数据存储和业务逻辑，与数据库交互
View（视图）：负责界面展示，将数据渲染给用户
Controller（控制器）：负责接收用户请求，调用 Model 获取数据，再交给 View 渲染

用户请求 → Controller → Model → Controller → View → 用户

Django 的 MTV 模式

Django 采用 MTV（Model-Template-View）模式：

Model（模型）：数据层，处理数据相关的一切，定义数据模型和数据库操作
Template（模板）：表现层，负责页面展示（HTML 模板）
View（视图）：业务逻辑层，接收请求、处理业务逻辑、返回响应

用户请求 → URL路由(urls.py) → View(views.py) → Model → View → Template → 用户

对应关系

MVC	Django MTV	说明
Model	Model	基本一致，都是数据层
View	Template	Django 的 Template 对应 MVC 的 View（界面展示）
Controller	View + URLconf	Django 的 View 函数/类承担了 Controller 的角色，URL 配置负责路由分发

为什么叫 MTV 而不叫 MVC？

Django 官方的说法是："在 Django 里，View 描述的是展示哪些数据，而 Template 描述的是数据如何展示。" 也就是说，Django 的 View 更像是 MVC 中的 Controller，而 Template 才是真正的 View。

本质上，MTV 和 MVC 是同一个东西的不同划分方式，都是为了实现关注点分离（Separation of Concerns），只是命名和职责划分略有不同。

Django 的完整请求流程

浏览器 → WSGI/ASGI Server
    → 中间件（请求阶段）
    → URL 路由（urls.py 匹配）
    → View（业务逻辑，操作 Model）
    → Template（渲染 HTML）
    → 中间件（响应阶段）
    → 浏览器

追问延伸：

Django 的 URL 路由是怎么匹配的？有哪些匹配方式？
除了 MTV，你还知道哪些 Web 架构模式？MVVM 和 MVC 有什么区别？
Django 的 View 有函数视图和类视图，各有什么优缺点？

Q3：Django 的请求处理流程？中间件机制？「🟡 中级」

考察点：考察对 Django 核心机制的理解深度，是否真正理解请求的完整生命周期，筛选只会写视图函数而不懂底层原理的人。

参考答案：

Django 请求处理完整流程

┌─────────────┐
│   浏览器     │
└──────┬──────┘
       │ HTTP 请求
       ▼
┌─────────────┐
│  Nginx等    │  反向代理（生产环境）
└──────┬──────┘
       ▼
┌─────────────┐
│ WSGI Server │  Gunicorn / uWSGI
└──────┬──────┘
       ▼
┌─────────────┐
│  中间件      │  process_request 从上到下依次执行
│  (请求阶段)   │
└──────┬──────┘
       ▼
┌─────────────┐
│ URL Router  │  urls.py 正则/路径匹配
└──────┬──────┘
       ▼
┌─────────────┐
│    View     │  视图函数/类，业务逻辑
│    Model    │  数据库操作
│  Template   │  模板渲染
└──────┬──────┘
       ▼
┌─────────────┐
│  中间件      │  process_response 从下到上依次执行
│  (响应阶段)   │
└──────┬──────┘
       ▼
┌─────────────┐
│   浏览器     │
└─────────────┘

中间件机制

中间件（Middleware）是 Django 的一个钩子框架，用于在请求和响应的处理过程中插入自定义逻辑，类似于 AOP（面向切面编程）。

中间件的五个钩子方法

方法	执行时机	说明
`process_request(request)`	请求到达 View 之前	返回 None 继续，返回 HttpResponse 直接短路
`process_view(request, view_func, view_args, view_kwargs)`	URL 匹配后，View 执行前	返回 None 继续，返回 HttpResponse 直接短路
`process_exception(request, exception)`	View 抛出异常时	返回 None 继续，返回 HttpResponse 作为响应
`process_response(request, response)`	响应返回给浏览器前	必须返回 HttpResponse 对象
`process_template_response(request, response)`	Template 渲染前	仅对 TemplateResponse 有效

执行顺序

中间件的执行顺序很重要：

请求方向：按 MIDDLEWARE 列表从上到下执行 process_request → process_view
响应方向：按 MIDDLEWARE 列表从下到上执行 process_response
异常方向：按 MIDDLEWARE 列表从下到上执行 process_exception

Middleware 1  ──► process_request ────────────────────────►
Middleware 2  ──► process_request ────────────────────────►
Middleware 3  ──► process_request ────────────────────────►
                     │
                     ▼
                  URL 路由
                     │
                     ▼
Middleware 1  ──► process_view ───────────────────────────►
Middleware 2  ──► process_view ───────────────────────────►
Middleware 3  ──► process_view ───────────────────────────►
                     │
                     ▼
                View 处理
                     │
         ┌───────────┴───────────┐
         ▼                       ▼
      正常返回                  异常
         │                       │
Middleware 3  ◄── process_response ◄────── process_exception ◄──┘
Middleware 2  ◄── process_response ◄────── process_exception ◄──┘
Middleware 1  ◄── process_response ◄────── process_exception ◄──┘
         │
         ▼
      返回响应

常用内置中间件

中间件	作用
`SecurityMiddleware`	安全相关，如 HTTPS 强制、XSS 防护等
`SessionMiddleware`	会话管理
`CsrfViewMiddleware`	CSRF 防护
`AuthenticationMiddleware`	用户认证，给 request 添加 user
`MessageMiddleware`	消息框架
`GZipMiddleware`	Gzip 压缩响应
`CommonMiddleware`	通用处理，如 URL 规范化、404 处理

自定义中间件示例

python

# myapp/middleware.py
import time
import logging

logger = logging.getLogger(__name__)

class RequestLoggingMiddleware:
    """记录请求耗时的中间件"""

    def __init__(self, get_response):
        # 初始化时调用一次，接收下一个中间件/视图的 callable
        self.get_response = get_response

    def __call__(self, request):
        # 请求进入
        start_time = time.time()
        request.start_time = start_time

        # 调用下一个中间件/视图
        response = self.get_response(request)

        # 响应返回
        duration = time.time() - start_time
        logger.info(
            f"{request.method} {request.path} - "
            f"状态码: {response.status_code} - "
            f"耗时: {duration:.3f}s"
        )
        response['X-Duration'] = f"{duration:.3f}s"

        return response

    def process_view(self, request, view_func, view_args, view_kwargs):
        # 在视图执行前执行
        logger.debug(f"即将执行视图函数: {view_func.__name__}")
        return None  # 返回 None 继续执行

    def process_exception(self, request, exception):
        # 视图抛出异常时执行
        logger.error(f"请求异常: {exception}", exc_info=True)
        return None  # 返回 None 继续抛出异常

注册中间件：

python

# settings.py
MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
    'myapp.middleware.RequestLoggingMiddleware',  # 自定义中间件
]

关键要点

中间件是全局的：每个请求都会经过所有中间件，所以中间件里不要做太重的操作
短路机制：process_request 和 process_view 返回 HttpResponse 时，会直接跳过后续中间件和 View，从当前位置反向执行 process_response
异常处理：process_exception 只有在 View 抛出异常时才会调用，且从下往上执行
Django 新版风格：推荐使用 __init__ + __call__ 的方式，更灵活

追问延伸：

如果我想在所有响应头里加一个自定义字段，应该怎么实现？
中间件和装饰器有什么区别？什么时候用中间件，什么时候用装饰器？
中间件的 process_request 返回一个 HttpResponse 会发生什么？
Django 的 CSRF 中间件是怎么工作的？
你在项目中写过哪些自定义中间件？解决了什么问题？

考察点：考察对 Django ORM 核心机制的理解，尤其是 N+1 问题和优化手段，筛选只会写 ORM 而不懂 SQL 和性能优化的人。

参考答案：

ORM 基本原理

ORM（Object-Relational Mapping，对象关系映射）是一种将面向对象语言中的对象与关系数据库中的表建立映射关系的技术。

Python 类  ←→  数据库表
Python 对象 ←→  表中的行（记录）
类属性     ←→  表字段

Django ORM 的核心概念：

Model：Python 类，继承 django.db.models.Model，对应数据库中的一张表
QuerySet：查询结果集，是惰性的（Lazy），链式调用，只有在真正需要数据时才执行 SQL
Manager：模型的查询接口，objects 是默认 Manager
Field：字段类型，映射到数据库列类型

QuerySet 的惰性查询

QuerySet 是惰性的，创建 QuerySet 不会立即执行 SQL，只有在以下情况才会真正查询数据库：

python

# 以下操作都不会执行 SQL
queryset = Book.objects.filter(price__gt=100)
queryset = queryset.order_by('-publish_date')
queryset = queryset.select_related('author')

# 以下操作才会执行 SQL
print(queryset)          # 迭代/打印
list(queryset)           # 转列表
for book in queryset:    # 遍历
    print(book.title)
queryset[0]              # 索引访问
len(queryset)            # 取长度（但会执行 count 优化）
bool(queryset)           # 布尔判断
queryset.exists()        # exists()

为什么要惰性查询？

可以链式调用，逐步构建查询条件
只有真正需要数据时才查询，避免不必要的数据库访问
给了 Django 优化查询的空间

N+1 查询问题

N+1 问题是 ORM 中最常见的性能陷阱：

python

# 查询所有书籍（1 次查询）
books = Book.objects.all()

for book in books:
    # 每本书访问作者，又执行一次查询（N 次查询）
    print(book.author.name)

# 总共执行了 1 + N 次查询

如果有 100 本书，就会执行 101 次 SQL 查询，性能极差。

Django 提供了两种方式解决 N+1 问题：

原理：通过 SQL 的 JOIN 操作，在一次查询中把关联对象的数据一起查出来
适用关系：一对一（OneToOneField）、外键（ForeignKey）—— 即"多对一"的"一"端
SQL 数量：1 次查询
限制：只能查前向的外键和一对一，不能查反向的一对多和多对多

python

# 一次查询，用 JOIN 查出所有书籍及其作者
books = Book.objects.select_related('author').all()

for book in books:
    # 不会再查询数据库，数据已经在内存中了
    print(book.author.name)

# 只执行了 1 次 SQL

生成的 SQL 大概是：

sql

SELECT book.*, author.* FROM book
INNER JOIN author ON book.author_id = author.id

原理：先查主表，再查关联表，然后在 Python 内存中做关联
适用关系：一对多（反向外键）、多对多（ManyToManyField）
SQL 数量：2 次查询（主表 1 次 + 关联表 1 次）
优势：可以查一对多和多对多，还可以通过 Prefetch 对象进行过滤

python

# 查询所有作者及其出版的书籍
authors = Author.objects.prefetch_related('books').all()

for author in authors:
    # 不会再查询数据库
    print(author.books.all())

# 执行了 2 次 SQL

生成的 SQL 大概是：

sql

SELECT * FROM author;
SELECT * FROM book WHERE author_id IN (1, 2, 3, ...);

然后 Django 在 Python 中把 books 按 author_id 分组，关联到对应的 author 对象上。

对比总结

维度	select_related	prefetch_related
原理	SQL JOIN 一次查出	两次查询，Python 中关联
SQL 次数	1 次	2 次（或更多，取决于层数）
适用关系	外键、一对一（多对一、一对一）	一对多、多对多
方向	只能正向（从有外键的一侧查）	正向反向都可以
性能	关联少时更快（一次 JOIN）	关联数据多时可能更快（避免大结果集）
支持过滤	不支持	支持（通过 Prefetch 对象）

进阶：Prefetch 对象

prefetch_related 可以配合 Prefetch 对象实现更复杂的预取：

python

from django.db.models import Prefetch

# 预取每本书的评论，但只取已审核的评论
books = Book.objects.prefetch_related(
    Prefetch(
        'comments',
        queryset=Comment.objects.filter(is_approved=True),
        to_attr='approved_comments'  # 存到自定义属性名
    )
).all()

for book in books:
    # 只包含已审核的评论
    print(book.approved_comments)

其他 ORM 优化技巧

python

# 1. only() / defer() - 只查询需要的字段，减少数据传输
books = Book.objects.only('title', 'price')  # 只查 title 和 price
books = Book.objects.defer('content')        # 排除 content 字段

# 2. values() / values_list() - 返回字典/元组，不构造 Model 对象
# 性能更好，适合只需要数据不需要对象方法的场景
book_titles = Book.objects.values_list('title', flat=True)

# 3. bulk_create / bulk_update - 批量操作
Book.objects.bulk_create([Book(title='...'), Book(title='...')])

# 4. annotate / aggregate - 聚合查询
from django.db.models import Count, Avg
Author.objects.annotate(book_count=Count('books'))

# 5. F() 表达式 - 在数据库层面做运算，避免拉取到 Python
from django.db.models import F
Book.objects.filter(price__lt=F('cost') * 2)

# 6. exists() / count() - 比 len() / bool() 更高效
if Book.objects.filter(title='Python').exists():
    ...

追问延伸：

Django ORM 怎么执行原生 SQL？有哪些方式？
什么情况下 select_related 反而会更慢？
prefetch_related 可以嵌套使用吗？如何实现多层预取？
Django 的 QuerySet 是线程安全的吗？
你在项目中遇到过最严重的 ORM 性能问题是什么？怎么解决的？
如何查看 Django ORM 生成的 SQL？

Q5：Django 的信号（Signal）是什么？应用场景？「🟡 中级」

考察点：考察对 Django 解耦机制的理解，是否知道如何在不侵入代码的情况下扩展功能，筛选只会写业务逻辑不懂架构解耦的人。

参考答案：

信号是什么？

信号（Signal）是 Django 提供的一种观察者模式（发布-订阅模式）的实现，用于在框架或应用的某些事件发生时，通知其他组件做出响应，实现了事件发送者和接收者之间的解耦。

发送者 ──发信号──► 信号中心 ──通知──► 接收者 A
                            ──通知──► 接收者 B
                            ──通知──► 接收者 C

核心思想：发送者不知道也不关心谁在监听，接收者可以独立注册和取消注册。

Django 内置信号

Django 提供了很多内置信号：

信号	触发时机	说明
`pre_save`	Model 保存前	可以修改实例数据
`post_save`	Model 保存后	常用于日志、缓存更新
`pre_delete`	Model 删除前
`post_delete`	Model 删除后
`m2m_changed`	多对多关系改变时
`request_started`	请求开始时
`request_finished`	请求结束时
`got_request_exception`	请求异常时

信号的使用

方式一：使用 receiver 装饰器

python

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth.models import User
from .models import UserProfile

@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
    """用户创建后自动创建用户资料"""
    if created:
        UserProfile.objects.create(user=instance)

@receiver(post_save, sender=User)
def save_user_profile(sender, instance, **kwargs):
    """用户保存时同步保存用户资料"""
    instance.profile.save()

方式二：使用 connect 方法

python

from django.db.models.signals import post_save

def my_handler(sender, **kwargs):
    print(f"{sender.__name__} 被保存了")

post_save.connect(my_handler, sender=Book)

自定义信号

除了使用内置信号，还可以定义自己的信号：

python

# myapp/signals.py
import django.dispatch

# 定义信号
order_created = django.dispatch.Signal()

# 发送信号
def create_order(user, items):
    order = Order.objects.create(user=user)
    # ... 创建订单逻辑 ...

    # 发送信号，传递参数
    order_created.send(
        sender=Order,
        order=order,
        user=user,
        items=items
    )
    return order

# 接收信号
@receiver(order_created, sender=Order)
def send_order_notification(sender, order, user, **kwargs):
    """发送订单通知邮件"""
    send_mail(
        subject='订单创建成功',
        message=f'您的订单 {order.id} 已创建',
        recipient_list=[user.email]
    )

@receiver(order_created, sender=Order)
def update_inventory(sender, items, **kwargs):
    """扣减库存"""
    for item in items:
        item.product.stock -= item.quantity
        item.product.save()

应用场景

日志记录：数据变更时自动记录审计日志
缓存更新：数据变化时使缓存失效
通知推送：用户注册后发送欢迎邮件、短信
数据同步：主数据变更后同步到其他系统/表
事件驱动：订单创建后触发库存扣减、积分增加等
扩展第三方应用：不修改第三方代码的情况下扩展功能

注意事项和坑

同步执行：信号是同步执行的，接收函数的执行时间会影响发送者的响应时间。不要在信号里做耗时操作（如发邮件、调外部接口），应该用 Celery 异步处理。

python

# ❌ 错误：同步发邮件，会拖慢用户注册速度
@receiver(post_save, sender=User)
def send_welcome_email(sender, instance, created, **kwargs):
    if created:
        send_mail(...)  # 同步阻塞

# ✅ 正确：用 Celery 异步发送
@receiver(post_save, sender=User)
def send_welcome_email(sender, instance, created, **kwargs):
    if created:
        send_welcome_email_task.delay(instance.id)

异常传播：信号接收函数抛出异常会影响发送者，导致整个操作失败。需要谨慎处理异常。
执行顺序不保证：多个接收者的执行顺序是不确定的，不要依赖顺序。
不要过度使用：信号会让代码流程变得不直观，增加调试难度。能用直接调用就用直接调用，信号是最后的选择。
断开信号：测试时可能需要断开信号，用 Signal.disconnect()。
弱引用：Django 默认使用弱引用存储接收者，如果接收者是局部变量可能会被垃圾回收。可以用 weak=False 解决。

信号 vs 直接调用 vs 中间件

方式	适用场景	耦合度	调试难度
直接调用	明确知道要调用谁	高	低
信号	多个接收者、解耦、扩展第三方	低	中
中间件	全局请求/响应处理	中	中

追问延伸：

信号是同步还是异步的？如果我想异步处理怎么办？
如果信号接收函数抛出异常会怎样？
你在项目中用过自定义信号吗？解决了什么问题？
Django 的 post_save 信号在 bulk_create 时会触发吗？为什么？
信号和观察者模式是什么关系？
除了信号，还有哪些解耦手段？

Q6：Flask 的上下文机制？请求上下文和应用上下文？「🟡 中级」

考察点：考察对 Flask 核心设计的理解深度，是否理解为什么 Flask 可以在视图中直接使用 request 等"全局"变量，筛选只会用 API 不懂原理的人。

参考答案：

为什么需要上下文？

在 Flask 中，我们可以在视图函数里直接使用 request、current_app 等变量，看起来像是全局变量：

python

from flask import Flask, request, current_app

app = Flask(__name__)

@app.route('/')
def index():
    # 直接用 request，不需要作为参数传入
    user_agent = request.headers.get('User-Agent')
    app_name = current_app.name
    return f'Hello, app={app_name}, ua={user_agent}'

但 request 不可能是真正的全局变量——因为 Web 服务器是多线程的，每个请求都有自己的 request 对象。如果是全局变量，多个请求就会互相覆盖。

Flask 的解决方案是上下文本地变量（Context Locals）：把变量存在线程（或协程）本地存储中，每个线程有自己独立的副本，看起来像全局变量，实际是线程隔离的。

两种上下文

Flask 有两种上下文：

1. 请求上下文（Request Context）

包含请求相关的信息，每次请求创建一个，请求结束销毁。

变量	说明
`request`	当前请求对象，包含请求的所有信息（方法、参数、头、体等）
`session`	当前会话，用于在多次请求间存储用户信息

2. 应用上下文（App Context）

包含应用相关的信息，不一定伴随请求存在。

变量	说明
`current_app`	当前处理请求的应用实例（Flask app 对象）
`g`	临时存储变量的容器，每个请求会重置，可用于在请求中传递数据

上下文栈

Flask 内部用栈来管理上下文，支持多应用场景：

_request_ctx_stack：请求上下文栈
_app_ctx_stack：应用上下文栈

请求上下文栈        应用上下文栈
┌──────────┐      ┌──────────┐
│ Request2 │      │   App2   │  ← 栈顶（当前）
├──────────┤      ├──────────┤
│ Request1 │      │   App1   │
└──────────┘      └──────────┘

当请求进来时，Flask 会：

创建请求上下文，压入请求上下文栈
如果应用上下文栈为空，创建应用上下文并压入
处理请求
弹出请求上下文
如果应用上下文是这次请求创建的，也弹出

为什么需要两种上下文？

既然有了请求上下文，为什么还要单独的应用上下文？主要原因：

多应用场景：一个进程中可以有多个 Flask 应用，每个应用有自己的配置。current_app 要能正确指向当前处理请求的那个应用。

离线脚本 / 测试：有些场景没有 HTTP 请求，但需要访问应用的配置和资源。比如：

python

# 离线脚本中手动推送应用上下文
from myapp import app

with app.app_context():
    # 现在可以使用 current_app、g 了
    db.create_all()
    print(current_app.config['DEBUG'])

解耦：应用上下文不依赖请求上下文，可以独立存在。比如在 Celery 任务中也需要访问应用配置。

LocalProxy 原理

Flask 的 request、current_app 等都是 LocalProxy 对象，代理了对上下文栈顶对象的属性访问。

核心原理基于 Werkzeug 的 Local：

python

# 简化版原理
from werkzeug.local import Local, LocalProxy

# 线程本地存储
_local = Local()

def _get_request():
    return _local.stack[-1]

# 代理对象，每次访问都会调用 _get_request 获取当前的 request
request = LocalProxy(_get_request)

每次访问 request.method 时，实际上是：

LocalProxy.__getattr__ 被调用
调用 _get_request() 从当前线程的栈顶获取真正的 request 对象
转发属性访问到真实对象上

这样，虽然 request 是全局导入的，但每次访问都会动态获取当前线程/协程的 request 对象。

上下文的生命周期

请求到达
    │
    ▼
创建 RequestContext
    │
    ▼
推送请求上下文到栈
    │
    ├──► 如果没有应用上下文，创建并推送
    │
    ▼
执行视图函数（可以使用 request、session、current_app、g）
    │
    ▼
弹出请求上下文
    │
    ├──► 如果应用上下文是这次创建的，也弹出
    │
    ▼
请求结束

手动使用上下文

在视图函数外（如测试、脚本）可以手动推送上下文：

python

# 1. 推送请求上下文
from flask import Flask, request

app = Flask(__name__)

with app.test_request_context('/hello?name=world', method='POST'):
    # 现在可以像在视图函数中一样使用 request
    assert request.path == '/hello'
    assert request.method == 'POST'
    assert request.args.get('name') == 'world'

# 2. 推送应用上下文
with app.app_context():
    assert current_app is app
    db.create_all()  # 数据库操作需要应用上下文
    g.user = 'admin'

# 3. before_request / after_request 也在请求上下文中

g 对象的使用

g 是应用上下文中的一个对象，用于在请求处理过程中存储临时数据：

python

from flask import g

@app.before_request
def load_user():
    user_id = session.get('user_id')
    if user_id:
        g.user = User.query.get(user_id)
    else:
        g.user = None

@app.route('/profile')
def profile():
    if g.user:
        return render_template('profile.html', user=g.user)
    return redirect(url_for('login'))

g vs session：

g 只在一次请求中有效，请求结束就重置
session 跨请求持久化，存在 Cookie 中

常见问题

Q：为什么会出现 "Working outside of application context" 错误？ A：因为在应用上下文外访问了 current_app 或 g。需要用 app.app_context() 推送上下文。

Q：为什么会出现 "Working outside of request context" 错误？ A：因为在请求上下文外访问了 request 或 session。需要在视图函数内或使用 test_request_context。

追问延伸：

Flask 的上下文是线程安全的吗？协程（async）下呢？
多应用场景下，current_app 怎么知道指向哪个 app？
g 对象和 session 有什么区别？
为什么 Django 没有类似的上下文机制，而是把 request 作为参数传给视图？
Werkzeug 的 Local 和 Python 的 threading.local 有什么区别？
你用过 Flask 的 g 对象吗？在什么场景下使用？

Q7：Flask 的蓝图（Blueprint）是什么？有什么用？「🟢 校招/初级」

考察点：考察对 Flask 模块化机制的理解，是否能组织中大型 Flask 项目，筛选只会写单文件应用的人。

参考答案：

蓝图是什么？

蓝图（Blueprint）是 Flask 提供的模块化应用组件，可以把路由、模板、静态文件等组织成独立的模块，然后注册到主应用上。

蓝图可以理解为"应用的半成品"——它不能独立运行，必须注册到一个 Flask 应用上才能生效。

为什么需要蓝图？

如果不用蓝图，一个 Flask 应用的所有路由都写在一个文件里：

python

# app.py - 所有代码堆在一起，维护困难
from flask import Flask

app = Flask(__name__)

@app.route('/')
def index(): ...

@app.route('/admin/users')
def admin_users(): ...

@app.route('/api/posts')
def api_posts(): ...

@app.route('/auth/login')
def auth_login(): ...
# ... 越来越多，几百上千行

用蓝图后，可以按业务模块拆分：

myapp/
  __init__.py          # 创建 app，注册蓝图
  auth/                # 认证模块
    __init__.py
    views.py           # auth_bp
  admin/               # 管理后台模块
    __init__.py
    views.py           # admin_bp
  api/                 # API 模块
    __init__.py
    views.py           # api_bp
  blog/                # 博客模块
    __init__.py
    views.py           # blog_bp

蓝图的使用

创建蓝图

python

# auth/views.py
from flask import Blueprint, render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, login_required

# 创建蓝图对象
auth_bp = Blueprint('auth', __name__)

# 在蓝图上定义路由
@auth_bp.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        # ... 登录逻辑 ...
        login_user(user)
        return redirect(url_for('main.index'))
    return render_template('auth/login.html')

@auth_bp.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect(url_for('main.index'))

注册蓝图

python

# __init__.py
from flask import Flask
from .auth.views import auth_bp
from .admin.views import admin_bp
from .api.views import api_bp

def create_app():
    app = Flask(__name__)

    # 注册蓝图，可以指定 URL 前缀
    app.register_blueprint(auth_bp, url_prefix='/auth')
    app.register_blueprint(admin_bp, url_prefix='/admin')
    app.register_blueprint(api_bp, url_prefix='/api/v1')

    return app

注册后，路由就变成了：

/auth/login
/auth/logout
/admin/...
/api/v1/...

蓝图的作用

模块化组织代码：按业务功能拆分，每个模块独立管理自己的路由、模板、静态文件
URL 前缀：给一组路由统一添加前缀，便于管理和版本控制
子域名支持：蓝图可以绑定到子域名，如 api.example.com
可复用组件：蓝图可以在多个应用间复用，比如通用的认证模块
模板/静态文件命名空间：蓝图可以有自己的 templates 和 static 目录

蓝图的模板和静态文件

python

# 蓝图可以有自己的模板目录和静态文件目录
auth_bp = Blueprint(
    'auth',
    __name__,
    template_folder='templates',   # 蓝图内的 templates 目录
    static_folder='static',       # 蓝图内的 static 目录
    static_url_path='/auth/static'
)

模板查找时，Flask 先找应用级的 templates，再找蓝图的 templates。

url_for 与蓝图

在蓝图中使用 url_for 时，端点名称要加上蓝图名前缀：

python

# 在蓝图内部，可以用 . 表示当前蓝图
@auth_bp.route('/login')
def login():
    # 两种方式都可以
    url_for('auth.login')    # 完整名称
    url_for('.login')        # 相对当前蓝图

# 在蓝图外部
url_for('auth.login')  # 必须用完整名称

蓝图 vs Django 的 app

特性	Flask Blueprint	Django App
模块化	是	是
独立路由	是	是（自己的 urls.py）
独立模型	否（模型还是全局的）	是（每个 app 有自己的 models）
独立模板/静态	是	是
可独立迁移	否	是（migrations）
Admin 集成	需自己实现	内置

蓝图的局限性

不能有自己的配置：蓝图共享应用的配置
不能独立运行：必须注册到应用上
没有独立的数据库模型：模型还是全局的，蓝图不负责管理模型

追问延伸：

蓝图可以嵌套吗？（蓝图里注册蓝图）
蓝图的 before_request 和 app 的 before_request 执行顺序是怎样的？
你在项目中是怎么用蓝图组织代码的？有哪些模块？
蓝图和 Flask 的 app 工厂模式有什么关系？
如果两个蓝图有同名的路由端点会怎样？

Q8：FastAPI 的特点和原理？为什么性能高？「🟡 中级」

考察点：考察对现代 Python Web 框架的理解，是否了解 ASGI、异步编程和类型提示等核心概念，筛选只懂传统 WSGI 框架的人。

参考答案：

FastAPI 是什么？

FastAPI 是一个现代、高性能的 Python Web 框架，用于构建 API，基于标准 Python 类型提示。它由 Sebastián Ramírez（tiangolo）创建，是目前 Python 生态中增长最快的 Web 框架之一。

FastAPI 的核心特点

1. 类型提示驱动（Type Hints Driven）

FastAPI 最大的特点是利用 Python 的类型提示（Type Hints）来自动完成参数验证、序列化和文档生成：

python

from fastapi import FastAPI, Path, Query
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool | None = None  # 可选字段

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., ge=1),        # 路径参数，自动验证 >= 1
    q: str | None = Query(None, max_length=50),  # 查询参数，自动验证长度
):
    return {"item_id": item_id, "q": q}

@app.post("/items/")
async def create_item(item: Item):  # 请求体自动验证
    return item

自动完成的工作：

参数类型转换（字符串自动转 int、float 等）
参数验证（范围、长度、格式等）
自动生成 JSON Schema
自动生成 API 文档
请求体解析
响应序列化

2. 自动生成 API 文档

启动应用后，访问：

/docs：Swagger UI 交互式文档
/redoc：ReDoc 文档

文档是自动从代码和类型提示生成的，代码即文档，永远不会过时。

3. 原生异步支持

FastAPI 基于 ASGI，原生支持 async/await：

python

import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/slow")
async def slow_endpoint():
    # 异步 I/O 操作，不会阻塞其他请求
    await asyncio.sleep(5)
    return {"message": "done"}

4. 高性能

FastAPI 的性能可以和 Node.js、Go 媲美（在 IO 密集型场景下），是性能最高的 Python Web 框架之一。

为什么 FastAPI 性能高？

1. 基于 ASGI 而非 WSGI

特性	WSGI	ASGI
模型	同步，一次处理一个请求	异步，原生支持并发
长连接	不支持	支持 WebSocket、SSE
调用方式	同步 callable	异步 callable
代表服务器	Gunicorn、uWSGI	Uvicorn、Daphne

WSGI 是同步的，每个请求占用一个线程/进程。ASGI 是异步的，基于事件循环，可以在单线程内处理大量并发请求（IO 等待时切换到其他请求）。

2. 基于 Starlette

Starlette 是一个轻量级的 ASGI 框架/工具包，本身性能就很高。FastAPI 在 Starlette 之上添加了数据验证、序列化、依赖注入等功能。

FastAPI  ←── 数据验证、依赖注入、自动文档
   │
   ▼
Starlette ←── ASGI 实现、路由、WebSocket、静态文件
   │
   ▼
ASGI Server (Uvicorn) ←── 异步服务器

3. 基于 Pydantic V2

Pydantic 是数据验证库，V2 版本用 Rust 重写了核心（pydantic-core），验证速度比 V1 快 5-50 倍。

FastAPI 的请求验证、响应序列化都依赖 Pydantic，这部分性能很高。

4. 依赖注入系统

FastAPI 的依赖注入（Depends）不仅提供了强大的代码组织能力，还做了很多优化：

python

from fastapi import Depends

async def get_db():
    db = Session()
    try:
        yield db
    finally:
        db.close()

@app.get("/items/")
async def read_items(db = Depends(get_db)):
    return db.query(Item).all()

FastAPI vs Flask vs Django 性能对比

根据 TechEmpower 等第三方基准测试（大致数量级）：

框架	每秒请求数（大致）	说明
FastAPI + Uvicorn	~15,000 - 50,000+	异步场景下最高
Flask + Gunicorn	~3,000 - 8,000	同步，多进程
Django + Gunicorn	~2,000 - 6,000	同步，功能重

注意：实际性能取决于具体场景（CPU 密集 vs IO 密集）、数据库查询、缓存等因素。框架本身的性能差异在实际业务中往往不是瓶颈。

FastAPI 的适用场景

高性能 API 服务：对响应时间和并发量要求高的场景
微服务：轻量、快速、文档完善，适合微服务架构
异步后端：需要处理大量并发连接、长连接的场景
机器学习 / AI 服务：快速将模型封装成 API
内部工具 / 原型：自动文档、类型提示，开发效率极高

FastAPI 的不足

生态不如 Django/Flask 成熟：第三方扩展相对较少
学习曲线：需要理解类型提示、异步编程、Pydantic 等概念
全栈能力弱：没有内置 ORM、Admin、模板引擎等，需要自己集成
社区相对年轻：中文资料和最佳实践相对较少

追问延伸：

FastAPI 的依赖注入系统是怎么工作的？和 Django/Flask 的中间件有什么区别？
异步框架一定比同步框架快吗？什么场景下同步反而更好？
FastAPI 怎么处理 WebSocket？
FastAPI 和 Starlette 是什么关系？可以只用 Starlette 吗？
你用过 FastAPI 做过什么项目？遇到了哪些坑？
FastAPI 的路径操作函数定义成 async 和不定义成 async 有什么区别？

Q9：Pydantic 的原理和使用场景？「🟡 中级」

考察点：考察对数据验证和类型系统的理解，是否了解现代 Python 开发中的数据建模方式，筛选只会用字典传数据的人。

参考答案：

Pydantic 是什么？

Pydantic 是一个 Python 数据验证和设置管理库，基于 Python 类型提示（Type Hints）。它可以让你用标准的 Python 类型注解来定义数据模型，自动完成验证、序列化和类型转换。

FastAPI 之所以强大，很大程度上是因为 Pydantic。

核心功能

1. 数据模型定义

python

from pydantic import BaseModel, Field
from datetime import datetime
from typing import List, Optional

class User(BaseModel):
    id: int
    name: str = Field(..., min_length=2, max_length=50)  # 必填，长度限制
    email: str = Field(..., pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')  # 正则验证
    age: int | None = Field(None, ge=0, le=150)  # 可选，范围限制
    tags: List[str] = []  # 列表，默认空列表
    created_at: datetime  # 自动解析字符串为 datetime

# 自动验证和类型转换
user = User(
    id='123',           # 字符串自动转 int
    name='Alice',
    email='alice@example.com',
    age=25,
    tags=['admin', 'user'],
    created_at='2024-01-01T12:00:00'  # 字符串自动转 datetime
)

print(user.id)  # 123 (int 类型)
print(user.created_at)  # 2024-01-01 12:00:00 (datetime 类型)

2. 数据验证

python

from pydantic import BaseModel, ValidationError, field_validator

class User(BaseModel):
    name: str
    password: str
    confirm_password: str

    # 自定义验证器
    @field_validator('name')
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError('姓名不能为空')
        return v.strip()

    # 跨字段验证
    @field_validator('confirm_password')
    @classmethod
    def passwords_match(cls, v: str, values) -> str:
        if 'password' in values.data and v != values.data['password']:
            raise ValueError('两次密码不一致')
        return v

# 验证失败时抛出异常
try:
    user = User(name='  ', password='123', confirm_password='456')
except ValidationError as e:
    print(e.errors())
    # [
    #     {'type': 'value_error', 'loc': ('name',), 'msg': '姓名不能为空', ...},
    #     {'type': 'value_error', 'loc': ('confirm_password',), 'msg': '两次密码不一致', ...}
    # ]

3. 序列化（导出）

python

user = User(id=1, name='Alice', email='alice@example.com')

# 转字典
user_dict = user.model_dump()  # {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}

# 转 JSON 字符串
user_json = user.model_dump_json()  # '{"id":1,"name":"Alice","email":"alice@example.com"}'

# 排除某些字段
user.model_dump(exclude={'email'})  # 不包含 email

# 只包含某些字段
user.model_dump(include={'id', 'name'})

# 排除 None 值
user.model_dump(exclude_none=True)

4. 嵌套模型

python

class Address(BaseModel):
    city: str
    street: str
    zip_code: str

class User(BaseModel):
    name: str
    address: Address  # 嵌套模型
    friends: List[Address]  # 嵌套模型列表

user = User(
    name='Alice',
    address={'city': 'Beijing', 'street': 'Main St', 'zip_code': '100000'},
    friends=[
        {'city': 'Shanghai', 'street': 'Nanjing Rd', 'zip_code': '200000'}
    ]
)

print(user.address.city)  # 'Beijing'

Pydantic V1 vs V2

特性	V1	V2
核心实现	纯 Python	Rust 核心（pydantic-core）
性能	基准线	快 5-50 倍
API	`BaseModel.dict()`	`BaseModel.model_dump()`
验证器	`@validator`	`@field_validator` / `@model_validator`
严格模式	支持有限	原生支持 strict mode
泛型	有限支持	完善的泛型支持
JSON Schema	支持	更完善的 JSON Schema 生成

V2 最大的变化是用 Rust 重写了核心验证逻辑（pydantic-core），性能大幅提升，同时 API 也有一些调整（方法名都加了 model_ 前缀）。

Pydantic 的原理

Pydantic 的核心工作流程：

输入数据（dict/JSON）
      │
      ▼
  类型解析（根据类型注解确定期望的类型）
      │
      ▼
  数据验证（类型检查、范围检查、自定义验证器）
      │
      ▼
  类型转换（如果需要且可能，如 str → int）
      │
      ▼
  构建模型实例（属性赋值）

关键技术点：

类型提示解析：通过 typing 模块和 inspect 模块解析类型注解
验证器链：每个字段有一系列验证器，按顺序执行
Rust 核心（V2）：验证逻辑在 Rust 中实现，通过 Python 绑定调用，速度极快
JSON Schema 生成：根据类型注解自动生成 JSON Schema

使用场景

API 请求/响应验证：FastAPI 中用于验证请求体、查询参数、路径参数，序列化响应
配置管理：Pydantic Settings（pydantic-settings）用于加载和验证环境变量、配置文件
数据清洗和转换：从外部系统（数据库、API、CSV）获取的数据进行验证和规范化
数据传输对象（DTO）：在不同层之间传递数据，保证数据结构的正确性
表单验证：Web 表单数据验证

配置管理示例

python

# pip install pydantic-settings
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    # 从环境变量读取，自动转换类型
    app_name: str = "MyApp"
    debug: bool = False
    port: int = 8000
    database_url: str
    redis_url: str | None = None

    # 配置来源
    model_config = SettingsConfigDict(
        env_file='.env',  # 从 .env 文件读取
        env_prefix='APP_'  # 环境变量前缀，如 APP_PORT
    )

settings = Settings()
print(settings.port)  # 从 APP_PORT 环境变量读取，自动转 int

Pydantic vs Dataclasses vs Attrs

特性	dataclass	attrs	pydantic
类型提示	是	是	是
数据验证	否	部分（需扩展）	是（核心功能）
类型转换	否	部分	是
JSON 序列化	需自己实现	需扩展	内置
JSON Schema	否	否	内置
性能	高	高	V2 很快，V1 稍慢

追问延伸：

Pydantic 的 model_validate() 和 model_validate_json() 有什么区别？
Pydantic 怎么处理可选字段和默认值？
Pydantic 的严格模式（strict mode）是什么？有什么用？
Pydantic 可以和 ORM 结合使用吗？（ORM 模式/from_attributes）
你在项目中除了 FastAPI 之外，还在哪些场景用过 Pydantic？
Pydantic 的验证器执行顺序是怎样的？

Q10：WSGI 和 ASGI 的区别？「🔴 高级」

考察点：考察对 Python Web 底层协议的深度理解，是否理解同步和异步 Web 模型的本质差异，筛选只会用框架不懂底层原理的人。

参考答案：

WSGI 是什么？

WSGI（Web Server Gateway Interface）是 Python Web 服务器和 Web 应用之间的标准接口，定义于 PEP 3333。

在 WSGI 出现之前，Python Web 框架各自为政，每个框架有自己的服务器接口，切换服务器或框架都很麻烦。WSGI 统一了接口，让任何兼容 WSGI 的服务器都可以运行任何兼容 WSGI 的框架。

WSGI 应用的形式

WSGI 应用是一个可调用对象（函数或类），接收两个参数：

python

def simple_app(environ, start_response):
    # environ: 字典，包含所有请求信息
    # start_response: 函数，用于发送响应状态和头部

    status = '200 OK'
    headers = [('Content-type', 'text/plain; charset=utf-8')]
    start_response(status, headers)

    # 返回可迭代对象，作为响应体
    return [b"Hello, World!"]

WSGI 的特点

同步阻塞：一个请求处理完才能处理下一个
简单：接口非常简单，容易实现
成熟：生态完善，服务器和框架都广泛支持
单请求模型：一次只处理一个请求，不支持长连接

WSGI 生态

类别	代表
服务器	Gunicorn、uWSGI、mod_wsgi、wsgiref（标准库）
框架	Django、Flask、Bottle、Pyramid
中间件	可以在 WSGI 应用之间包装，形成中间件链

ASGI 是什么？

ASGI（Asynchronous Server Gateway Interface）是 WSGI 的异步继任者，为 Python 异步 Web 提供标准接口。

ASGI 支持：

异步应用（async/await）
长连接（WebSocket、Server-Sent Events）
HTTP/2
后台任务

ASGI 应用的形式

ASGI 应用是一个异步可调用对象，接收三个参数：

python

async def app(scope, receive, send):
    # scope: 字典，包含连接信息（类似 WSGI 的 environ）
    # receive: 异步可调用，用于接收事件（如请求体）
    # send: 异步可调用，用于发送事件（如响应）

    assert scope['type'] == 'http'

    # 发送响应头
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [(b'content-type', b'text/plain')],
    })

    # 发送响应体
    await send({
        'type': 'http.response.body',
        'body': b'Hello, World!',
    })

ASGI 是事件驱动的，通过 receive 和 send 收发事件，而不是像 WSGI 那样直接返回响应体。这使得它可以支持长连接——可以持续收发事件。

ASGI 的三种协议类型

ASGI 支持多种协议，通过 scope['type'] 区分：

type	协议	说明
`http`	HTTP	普通 HTTP 请求
`websocket`	WebSocket	双向通信长连接
`lifespan`	生命周期	应用启动/关闭事件

ASGI 生态

类别	代表
服务器	Uvicorn、Daphne、Hypercorn
框架	FastAPI、Starlette、Django 3.1+、Quart、Sanic

WSGI vs ASGI 对比

维度	WSGI	ASGI
全称	Web Server Gateway Interface	Asynchronous Server Gateway Interface
出现时间	2003 年（PEP 333）	2019 年左右
编程模型	同步	异步（也支持同步）
调用方式	同步 callable，返回可迭代对象	异步 callable，通过 receive/send 收发事件
并发模型	多进程/多线程	单线程事件循环 + 异步
性能	中等（受限于进程/线程数）	高（IO 密集场景下并发量极大）
HTTP 长连接	不支持	支持（WebSocket、SSE）
HTTP/2	不支持	支持
后台任务	不支持	可实现
学习曲线	平缓	较陡（需要理解异步编程）
生态成熟度	非常成熟	快速发展中
兼容性	只能运行 WSGI 应用	可以运行 WSGI 应用（通过包装）

为什么 ASGI 性能更高？

关键在于IO 等待时的处理方式：

WSGI（同步）：

请求1 ────处理────► IO等待(阻塞) ────继续处理────► 响应
                                 请求2 ────等待────► 处理────► ...

每个请求占一个线程/进程，IO 等待时线程被阻塞，不能干别的事。并发量受限于线程/进程数。

ASGI（异步）：

请求1 ──处理──► IO等待(挂起) ────────────────► 继续处理──► 响应
                请求2 ──处理──► IO等待(挂起) ────────────────► 继续处理──► 响应

单线程内，IO 等待时挂起当前请求，切换去处理其他请求。一个线程可以处理成千上万个并发请求。

但这只适用于 IO 密集型场景。如果是 CPU 密集型任务，异步和同步的性能差不多（因为 CPU 不会空闲）。

ASGI 运行 WSGI 应用

ASGI 服务器可以通过包装层运行 WSGI 应用（但反过来不行）：

python

# WSGI 应用可以运行在 ASGI 服务器上
from a2wsgi import WSGIMiddleware
from my_wsgi_app import application  # Django/Flask 应用

asgi_app = WSGIMiddleware(application)

Uvicorn 等 ASGI 服务器也内置了 WSGI 支持。

选型建议

场景	推荐	原因
传统 Web 应用、CMS	Django + Gunicorn（WSGI）	生态成熟，开发效率高
小型 API、快速原型	Flask + Gunicorn（WSGI）	简单灵活
高性能 API、微服务	FastAPI + Uvicorn（ASGI）	性能高，开发体验好
WebSocket、实时通信	FastAPI/Starlette + Uvicorn（ASGI）	WSGI 不支持长连接
大量并发 IO	ASGI	异步并发优势明显
CPU 密集型计算	WSGI（多进程）	异步优势不明显，多进程更简单

常见误区

"ASGI 一定比 WSGI 快"：不一定。如果应用主要是 CPU 密集型，或者数据库是瓶颈，ASGI 的优势不明显。
"用了 async 就一定快"：如果异步代码里有同步阻塞操作（如同步的数据库驱动），还是会阻塞事件循环，反而更慢。
"Django 不能用 ASGI"：Django 3.1+ 支持异步视图和 ASGI，但大部分 Django 生态还是同步的。

追问延伸：

ASGI 的 scope 里包含哪些信息？和 WSGI 的 environ 有什么不同？
什么是事件循环？Python 的 asyncio 是怎么工作的？
异步框架中如何安全地执行 CPU 密集型任务？
Uvicorn 和 Gunicorn 是什么关系？为什么生产环境常用 gunicorn + uvicorn？
WebSocket 为什么不能用 WSGI 实现？
你在项目中遇到过异步的坑吗？怎么解决的？

Q11：Python Web 应用的部署方式有哪些？「🟡 中级」

考察点：考察对生产环境部署的实战经验，是否理解从开发到生产的完整流程，筛选只会写代码不会部署的人。

参考答案：

开发环境 vs 生产环境

开发环境和生产环境的部署方式完全不同：

环境	部署方式	特点
开发	`manage.py runserver` / `flask run` / `uvicorn`	单进程、热重载、仅开发用
生产	Gunicorn/Uvicorn + Nginx	多进程、高性能、稳定安全

注意：绝对不要用开发服务器跑生产环境——性能差、不安全、不稳定。

生产环境部署方案

方案一：Gunicorn（WSGI）+ Nginx（Django/Flask 标准部署）

最经典的 Python Web 部署方案：

┌──────────┐     ┌───────────┐     ┌──────────────────┐
│  浏览器   │────►│   Nginx   │────►│  Gunicorn        │
│          │     │  (反向代理)│     │  (多 Worker 进程)│
└──────────┘     └───────────┘     └────────┬─────────┘
                                             │
                                             ▼
                                      Django/Flask 应用
                                             │
                                             ▼
                                          数据库

各组件职责：

组件	职责
Nginx	反向代理、静态文件服务、负载均衡、HTTPS 终止、Gzip 压缩、限流、缓存
Gunicorn	WSGI 服务器，管理 Worker 进程，接收请求转给 Django/Flask
Django/Flask	业务逻辑处理

Gunicorn 配置示例：

bash

# 启动命令
gunicorn myapp.wsgi:application \
    --bind 127.0.0.1:8000 \
    --workers 4 \          # Worker 进程数，一般 CPU核数*2+1
    --worker-class sync \  # sync / gevent / eventlet
    --timeout 30 \
    --access-logfile /var/log/gunicorn/access.log \
    --error-logfile /var/log/gunicorn/error.log

Nginx 配置示例：

nginx

server {
    listen 80;
    server_name example.com;

    # 静态文件直接由 Nginx 处理，不经过 Python
    location /static/ {
        alias /var/www/myapp/static/;
        expires 30d;  # 缓存静态文件
    }

    location /media/ {
        alias /var/www/myapp/media/;
    }

    # 动态请求转发给 Gunicorn
    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

方案二：Uvicorn（ASGI）+ Nginx（FastAPI 部署）

FastAPI 等异步框架用 ASGI 服务器部署：

┌──────────┐     ┌───────────┐     ┌──────────────────┐
│  浏览器   │────►│   Nginx   │────►│  Uvicorn         │
│          │     │  (反向代理)│     │  (ASGI 服务器)    │
└──────────┘     └───────────┘     └────────┬─────────┘
                                             │
                                             ▼
                                      FastAPI 应用

生产环境推荐用 Gunicorn 管理 Uvicorn Worker：

bash

# 用 Gunicorn 启动 Uvicorn Worker
# Gunicorn 管理进程，Uvicorn 处理 ASGI
gunicorn main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 127.0.0.1:8000

这样可以利用 Gunicorn 成熟的进程管理能力，同时享受 Uvicorn 的异步处理能力。

方案三：Docker + 容器编排

┌─────────────────────────────┐
│        Docker 容器          │
│  ┌───────────────────────┐  │
│  │  Nginx / Gunicorn     │  │
│  │  + Python App         │  │
│  └───────────────────────┘  │
└─────────────────────────────┘

Dockerfile 示例：

dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["gunicorn", "myapp.wsgi:application", "--bind", "0.0.0.0:8000", "--workers", "4"]

编排方式：

Docker Compose：单机部署，适合中小型项目
Kubernetes (K8s)：集群部署，适合大型项目，支持自动扩缩容、滚动更新

方案四：Serverless（无服务器）

平台	说明
AWS Lambda	最成熟的 Serverless 平台
阿里云函数计算	国内常用
腾讯云 SCF	国内常用
Vercel / Netlify	前端 + Serverless 一体化

适合：流量波动大、请求不频繁的场景，按调用次数付费，免运维。

方案五：PaaS（平台即服务）

平台	说明
Heroku	最经典的 PaaS，简单易用
Railway	现代化 PaaS
Render	全栈托管
Fly.io	边缘部署
阿里云 SAE	国内 Serverless 应用引擎

适合：中小型团队，不想运维基础设施，专注业务开发。

Nginx 的作用

Nginx 在部署中扮演着多重角色：

反向代理：接收请求转发给后端应用
静态文件服务：直接处理静态资源（CSS/JS/图片），不经过 Python
负载均衡：多台应用服务器之间分发请求
HTTPS 终止：处理 SSL/TLS 加解密
Gzip/Brotli 压缩：压缩响应，减少传输量
缓存：缓存静态资源和部分动态内容
限流 / 防爬：控制请求速率
访问控制：IP 白名单、黑名单

Worker 进程数配置

经验公式：Worker 数 = CPU 核数 * 2 + 1

但实际要根据场景调整：

IO 密集型应用：可以适当增加 Worker 数，或使用 gevent/eventlet 异步 Worker
CPU 密集型应用：Worker 数不要超过 CPU 核数
内存限制：每个 Worker 占用一定内存，Worker 太多可能导致 OOM

进程模型对比

Worker 类型	并发模型	适用场景
sync	多进程同步	CPU 密集、简单可靠
gevent	多进程 + 协程	IO 密集、高并发
eventlet	多进程 + 协程	IO 密集、高并发
UvicornWorker	多进程 + asyncio	异步 ASGI 应用

部署最佳实践

静态文件分离：静态文件由 Nginx 或 CDN 处理，不经过 Python
日志管理：应用日志输出到文件或集中式日志系统（ELK、Loki）
进程管理：用 Systemd 或 Supervisor 管理 Gunicorn/Uvicorn 进程
健康检查：配置健康检查端点，监控应用状态
环境变量：敏感信息（数据库密码、密钥等）通过环境变量注入，不要硬编码
CI/CD：自动化构建和部署，减少人工操作
监控告警：接入 APM（New Relic、Datadog、SkyWalking）监控性能

追问延伸：

Gunicorn 的 sync worker 和 gevent worker 有什么区别？怎么选？
Nginx 怎么做负载均衡？有哪些负载均衡算法？
Docker 部署 Python 应用有哪些最佳实践？怎么减小镜像体积？
如何实现零停机部署（滚动更新）？
生产环境中如何排查 Python 应用的性能问题？
你部署过的最大规模的 Python 应用是什么样的架构？

Q12：SQLAlchemy 是什么？和 Django ORM 比有什么区别？「🟡 中级」

考察点：考察对 Python ORM 生态的全面了解，是否理解不同 ORM 的设计哲学和适用场景，筛选只知道 Django ORM 的人。

参考答案：

SQLAlchemy 是什么？

SQLAlchemy 是 Python 生态中最流行、功能最强大的 ORM（对象关系映射）库，被誉为"Python 数据库工具的瑞士军刀"。

SQLAlchemy 有两层架构：

┌──────────────────────────────────┐
│          SQLAlchemy ORM          │  高层 ORM，面向对象
├──────────────────────────────────┤
│         SQLAlchemy Core          │  底层 SQL 表达式语言
└──────────────────────────────────┘

Core（SQL 表达式语言）：提供了 Pythonic 的 SQL 构建方式，可以用 Python 代码写 SQL，比拼字符串安全且方便
ORM：在 Core 之上提供了完整的 ORM 功能，将 Python 类映射到数据库表

SQLAlchemy 的核心特点

1. 灵活性极强

可以在 ORM 和原生 SQL 之间无缝切换：

python

from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

engine = create_engine("mysql+pymysql://user:pass@localhost/db")

# 方式一：原生 SQL
with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM users WHERE id = :id"), {"id": 1})
    row = result.fetchone()

# 方式二：SQL 表达式语言（Core）
from sqlalchemy import select, Table, Column, Integer, String

users = Table('users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
)

stmt = select(users).where(users.c.id == 1)
result = conn.execute(stmt)

# 方式三：ORM
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]

with Session(engine) as session:
    user = session.get(User, 1)

2. 强大的查询能力

SQLAlchemy 的查询能力远超 Django ORM，几乎可以表达任何 SQL：

python

# 复杂查询示例
from sqlalchemy import func, and_, or_, case

# 子查询
subq = select(func.count(Order.id).label('order_count'), Order.user_id)\
    .group_by(Order.user_id).subquery()

# JOIN + 聚合 + 排序
stmt = select(User, subq.c.order_count)\
    .join(subq, User.id == subq.c.user_id)\
    .where(User.age > 18)\
    .order_by(subq.c.order_count.desc())\
    .limit(10)

# CASE WHEN
stmt = select(
    User.name,
    case(
        (User.age < 18, "未成年"),
        (User.age < 60, "成年"),
        else_="老年"
    ).label("age_group")
)

3. 异步支持

SQLAlchemy 1.4+ 支持 asyncio：

python

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")

async def get_user(user_id: int):
    async with AsyncSession(engine) as session:
        user = await session.get(User, user_id)
        return user

4. 框架无关

SQLAlchemy 是独立的库，可以和任何 Web 框架配合使用：

Flask + SQLAlchemy（Flask-SQLAlchemy）
FastAPI + SQLAlchemy
Django 也可以用 SQLAlchemy（但一般用 Django ORM）

SQLAlchemy vs Django ORM

维度	SQLAlchemy	Django ORM
设计哲学	灵活强大，给开发者最大控制权	简单易用，约定大于配置
学习曲线	陡峭	平缓
查询能力	极强，几乎支持所有 SQL	够用，但复杂查询需写原生 SQL
灵活性	极高，可以精细控制 SQL	中等，受限于 Django 的设计
异步支持	完善（asyncio）	部分支持（Django 4.1+）
框架绑定	独立，可配合任何框架	和 Django 深度绑定
Admin 后台	无（需第三方如 Flask-Admin）	内置，功能强大
迁移工具	Alembic	内置 migrations
生态集成	需要自己集成各种工具	内置 Auth、Session、Form 等
适合场景	复杂查询、高性能、非 Django 项目	Django 项目、快速开发

详细对比

查询对比

python

# Django ORM
users = User.objects.filter(age__gt=18).order_by('-created_at')[:10]

# SQLAlchemy ORM
stmt = select(User).where(User.age > 18).order_by(User.created_at.desc()).limit(10)
users = session.execute(stmt).scalars().all()

关联查询对比

python

# Django ORM - 类似 select_related
books = Book.objects.select_related('author').all()

# SQLAlchemy - joinedload（类似 select_related）
from sqlalchemy.orm import joinedload

stmt = select(Book).options(joinedload(Book.author))
books = session.execute(stmt).scalars().unique().all()

# Django ORM - 类似 prefetch_related
authors = Author.objects.prefetch_related('books').all()

# SQLAlchemy - selectinload（类似 prefetch_related）
from sqlalchemy.orm import selectinload

stmt = select(Author).options(selectinload(Author.books))
authors = session.execute(stmt).scalars().all()

原生 SQL 支持

python

# Django ORM
User.objects.raw('SELECT * FROM users WHERE age > %s', [18])
# 或
from django.db import connection
with connection.cursor() as cursor:
    cursor.execute('SELECT * FROM users WHERE age > %s', [18])

# SQLAlchemy
session.execute(text('SELECT * FROM users WHERE age > :age'), {'age': 18})

什么时候选 SQLAlchemy？

非 Django 项目：Flask、FastAPI 项目首选 SQLAlchemy
复杂查询需求：需要复杂 JOIN、子查询、窗口函数等
需要精细控制 SQL：对性能要求高，需要优化每一条 SQL
多数据库支持：需要连接多种数据库，SQLAlchemy 支持几乎所有关系型数据库
异步需求：需要异步数据库操作

什么时候选 Django ORM？

Django 项目：和 Django 深度集成，开发效率最高
快速开发：内置 Admin、Auth、迁移等工具，开箱即用
团队协作：统一的约定，降低沟通成本
业务不复杂：标准的 CRUD 操作，Django ORM 足够用

SQLAlchemy 版本演进

版本	关键变化
1.x	经典版本，`session.query()` 风格
1.4	引入 2.0 风格，支持 `select()` 语法，支持 asyncio
2.0	完全采用新 API，更简洁的类型提示支持（Mapped/mapped_column）

SQLAlchemy 2.0 是目前的主流版本，推荐新项目使用。

追问延伸：

SQLAlchemy 的 Session 是什么？和 Django 的 QuerySet 有什么区别？
SQLAlchemy 的 Unit of Work 模式是什么？
SQLAlchemy 的懒加载和预加载有哪些方式？（lazyload、joinedload、selectinload、subqueryload）
Alembic 是什么？和 Django migrations 比有什么区别？
SQLAlchemy 怎么处理数据库连接池？
你觉得 SQLAlchemy 比 Django ORM 难在哪里？有学习建议吗？
有没有在同一个项目中同时使用 Django ORM 和 SQLAlchemy？

Q13：Django 中如何处理高并发？有哪些优化手段？「🔴 高级」

考察点：考察对高并发场景下系统优化的综合能力，是否有实战经验和全局视野，筛选只会写 CRUD 不懂性能优化的人。

参考答案：

Django 本身是同步框架，在高并发场景下需要从多个层面进行优化。优化是一个系统性工程，需要从数据库、缓存、异步、部署等多个维度入手。

一、数据库优化（最常见的瓶颈）

1. 查询优化

python

# 1. 避免 N+1 查询
# ❌
books = Book.objects.all()
for book in books:
    print(book.author.name)  # 每次都查数据库

# ✅
books = Book.objects.select_related('author').all()  # 一次 JOIN 查询
# 或者
authors = Author.objects.prefetch_related('books').all()  # 两次查询

# 2. 只查询需要的字段
# ❌
users = User.objects.all()  # 查出所有字段

# ✅
users = User.objects.only('name', 'email')  # 只查需要的字段
users = User.objects.values('id', 'name')   # 返回字典，不构造对象

# 3. 使用索引
class Book(models.Model):
    title = models.CharField(max_length=200, db_index=True)  # 加索引
    isbn = models.CharField(max_length=20, unique=True)      # 唯一索引

    class Meta:
        indexes = [
            models.Index(fields=['author', 'publish_date']),  # 联合索引
        ]

# 4. 批量操作
# ❌
for book in books:
    book.price += 10
    book.save()  # N 次 UPDATE

# ✅
Book.objects.filter(id__in=book_ids).update(price=F('price') + 10)  # 1 次 UPDATE

2. 数据库连接池

Django 默认每次请求创建一个数据库连接，请求结束关闭。高并发下连接开销大。

解决方案：使用数据库连接池

PgBouncer（PostgreSQL）：连接池中间件
数据库配置：CONN_MAX_AGE 设置连接复用时间

python

# settings.py
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'mydb',
        'CONN_MAX_AGE': 60,  # 连接复用 60 秒
        'CONN_HEALTH_CHECKS': True,  # 连接健康检查
    }
}

3. 读写分离

读多写少的场景，使用主从架构：

python

# settings.py
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'primary_db',
        'HOST': 'primary.example.com',
    },
    'replica': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'replica_db',
        'HOST': 'replica.example.com',
    }
}

# 自定义数据库路由
class PrimaryReplicaRouter:
    def db_for_read(self, model, **hints):
        return 'replica'  # 读操作走从库

    def db_for_write(self, model, **hints):
        return 'default'  # 写操作走主库

DATABASE_ROUTERS = ['myapp.router.PrimaryReplicaRouter']

4. 分库分表

数据量极大时考虑分库分表，但这是最后的手段，引入复杂度很高。

二、缓存优化

1. Django 缓存框架

Django 内置缓存框架，支持多种后端：

python

# settings.py - Redis 缓存
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.redis.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/0',
        'TIMEOUT': 300,  # 默认缓存 5 分钟
        'OPTIONS': {
            'MAX_ENTRIES': 10000,
            'CULL_FREQUENCY': 3,
        }
    }
}

2. 多级缓存策略

浏览器缓存 → CDN → Nginx 缓存 → 应用缓存(Redis) → 数据库

3. 缓存粒度

缓存级别	适用场景	代码示例
页面缓存	整页静态内容	`@cache_page(60 * 15)`
片段缓存	页面部分内容	`{% cache 500 sidebar %}`
对象缓存	单个数据对象	`cache.get('user:123')`
查询缓存	数据库查询结果	手动缓存 QuerySet

python

# 对象缓存示例
from django.core.cache import cache

def get_user_profile(user_id):
    cache_key = f'user_profile:{user_id}'
    profile = cache.get(cache_key)
    if profile is None:
        profile = UserProfile.objects.get(user_id=user_id)
        cache.set(cache_key, profile, 300)  # 缓存 5 分钟
    return profile

# 数据更新时清除缓存
def update_user_profile(user_id, data):
    profile = UserProfile.objects.get(user_id=user_id)
    # ... 更新 ...
    cache.delete(f'user_profile:{user_id}')

三、异步化

1. Celery 异步任务

耗时操作（发邮件、数据处理、调用外部 API）不要在请求中同步执行：

python

# ❌ 同步发邮件，阻塞请求
def register(request):
    user = User.objects.create(...)
    send_welcome_email(user)  # 可能耗时几秒
    return redirect('/')

# ✅ 异步发送
from .tasks import send_welcome_email_task

def register(request):
    user = User.objects.create(...)
    send_welcome_email_task.delay(user.id)  # 立即返回，后台执行
    return redirect('/')

2. Django 异步视图

Django 3.1+ 支持异步视图：

python

async def async_view(request):
    # 异步 I/O 操作
    data = await async_http_client.get('https://api.example.com/data')
    return JsonResponse(data)

但注意：Django 的 ORM 大部分操作还是同步的，异步视图中调用 ORM 还是会阻塞事件循环。需要用 sync_to_async 包装，或者使用异步数据库驱动。

四、部署优化

1. 多 Worker + Gevent

bash

# 使用 gevent worker，提高并发能力
gunicorn myapp.wsgi:application \
    --workers 4 \
    --worker-class gevent \
    --worker-connections 1000 \
    --bind 0.0.0.0:8000

gevent 用协程实现高并发，适合 IO 密集型应用。

2. Nginx 优化

nginx

# 开启 gzip 压缩
gzip on;
gzip_types text/plain text/css application/json application/javascript;
gzip_min_length 1024;

# 静态文件缓存
location /static/ {
    alias /var/www/static/;
    expires 30d;
    add_header Cache-Control "public, immutable";
}

# 负载均衡
upstream django_backend {
    server 127.0.0.1:8000;
    server 127.0.0.1:8001;
    server 127.0.0.1:8002;
    keepalive 32;  # 长连接
}

3. 静态资源 CDN

静态文件（CSS、JS、图片）放到 CDN 上，减轻服务器压力，加快访问速度。

五、其他优化手段

手段	说明
CDN 加速	静态资源、动态内容加速
HTTP/2	多路复用，减少连接开销
消息队列削峰	突发流量用消息队列缓冲
代码级优化	使用更高效的算法、减少不必要的计算
Protobuf 替代 JSON	序列化更快，数据更小
连接池	数据库、Redis、HTTP 连接都用连接池

性能优化方法论

先测量再优化：不要凭感觉优化，用工具（Django Debug Toolbar、New Relic、Py-Spy）定位瓶颈
二八定律：80% 的性能问题出在 20% 的代码上，优先优化热点
数据库是首要瓶颈：大多数 Web 应用的瓶颈在数据库，先优化数据库
缓存是银弹：能用缓存解决的就用缓存
权衡取舍：优化是有成本的，不要过度优化，够用就行

性能监控工具

Django Debug Toolbar：开发环境查看 SQL、缓存等信息
Silk：Django 请求剖析工具
New Relic / Datadog / SkyWalking：生产环境 APM
py-spy：Python 性能采样分析
explain / analyze：数据库查询计划分析

追问延伸：

你们系统的 QPS 是多少？瓶颈在哪里？怎么优化的？
缓存击穿、缓存穿透、缓存雪崩是什么？怎么解决？
数据库读写分离后，主从延迟怎么处理？
消息队列怎么实现削峰填谷？
Django 的 GIL 会不会影响性能？怎么解决？
如果流量突然增加 10 倍，你的系统能扛住吗？怎么扩容？
你做过的最有成就感的性能优化是什么？提升了多少？

Q14：Celery 的原理和架构？「🟡 中级」

考察点：考察对异步任务队列的理解，是否有实战经验，筛选不知道怎么处理异步任务的人。

参考答案：

Celery 是什么？

Celery 是 Python 生态中最流行的分布式任务队列，用于处理异步任务和定时任务。它可以把耗时操作从主流程中剥离出来，后台异步执行，提高系统响应速度和吞吐量。

Celery 架构

Celery 由以下几个核心组件组成：

┌─────────────┐
│   Producer   │  生产者：Web 应用，发送任务
└──────┬──────┘
       │ 发送任务消息
       ▼
┌─────────────┐
│    Broker    │  消息中间件：存储任务消息
│  (RabbitMQ   │
│   / Redis)   │
└──────┬──────┘
       │ 消费任务
       ▼
┌─────────────┐
│    Worker    │  工作进程：执行任务
│  (进程/协程) │
└──────┬──────┘
       │ 存储结果
       ▼
┌─────────────┐
│   Backend    │  结果存储：保存任务执行结果
│  (Redis/DB   │
│   / AMQP)    │
└─────────────┘

┌─────────────┐
│     Beat     │  定时任务调度器：周期性发送任务
└──────┬──────┘
       │
       └──────► Broker

各组件详解

组件	作用	可选实现
Producer	产生任务，发送到 Broker	Web 应用、脚本、其他服务
Broker	消息中间件，存储任务队列	RabbitMQ（推荐）、Redis、SQS
Worker	从 Broker 取任务并执行	Celery Worker（多进程/协程）
Backend	存储任务执行结果和状态	Redis、数据库、AMQP、Memcached
Beat	定时任务调度器	Celery Beat（内置）

工作流程

1. 应用调用任务函数的 .delay() 或 .apply_async()
         │
         ▼
2. Celery 将任务序列化成消息，发送到 Broker
         │
         ▼
3. Broker 将消息存入对应的队列
         │
         ▼
4. Worker 从 Broker 的队列中获取任务
         │
         ▼
5. Worker 执行任务
         │
         ▼
6. 执行结果存储到 Backend（如果配置了）

快速上手

python

# tasks.py
from celery import Celery

# 创建 Celery 实例
app = Celery(
    'myapp',
    broker='redis://localhost:6379/0',   # Broker
    backend='redis://localhost:6379/1',  # 结果存储
)

# 定义任务
@app.task
def add(x, y):
    return x + y

@app.task
def send_welcome_email(user_id):
    """发送欢迎邮件"""
    user = User.objects.get(id=user_id)
    send_mail(
        subject='欢迎加入',
        message=f'你好，{user.name}！',
        from_email='noreply@example.com',
        recipient_list=[user.email]
    )
    return f'邮件已发送给 {user.email}'

发送任务：

python

# 方式一：delay() - 简单方式
result = add.delay(2, 3)

# 方式二：apply_async() - 更丰富的控制
result = add.apply_async(
    args=[2, 3],
    countdown=10,       # 10 秒后执行
    eta=datetime(...),  # 指定执行时间
    queue='high_priority',  # 指定队列
)

# 获取任务结果
print(result.id)           # 任务 ID
print(result.ready())      # 是否完成
print(result.get(timeout=1))  # 获取结果（阻塞）
print(result.state)        # 任务状态：PENDING / STARTED / SUCCESS / FAILURE

启动 Worker：

bash

# 启动 Worker
celery -A tasks worker --loglevel=info

# 启动定时任务调度器
celery -A tasks beat

Broker 选型

Broker	优点	缺点	适用场景
RabbitMQ	功能完善、可靠、支持消息确认	较重，维护成本高	生产环境、高可靠性要求
Redis	轻量、快速、部署简单	功能相对少，可靠性稍差	中小项目、快速开发
SQS	托管服务，免运维	AWS 绑定，有延迟	AWS 生态项目

生产环境推荐 RabbitMQ，开发和小项目可以用 Redis。

Worker 并发模型

模式	说明	适用场景
prefork	多进程（默认）	CPU 密集型任务
gevent	协程（基于 gevent）	IO 密集型任务，高并发
eventlet	协程（基于 eventlet）	IO 密集型任务
solo	单进程	调试用

bash

# 多进程模式，4 个 worker
celery -A tasks worker --concurrency=4

# gevent 协程模式，1000 个并发
celery -A tasks worker --pool=gevent --concurrency=1000

定时任务（Celery Beat）

python

# celery_config.py
from celery.schedules import crontab

app.conf.beat_schedule = {
    # 每天凌晨 3 点执行
    'cleanup-old-data': {
        'task': 'myapp.tasks.cleanup_old_data',
        'schedule': crontab(hour=3, minute=0),
        'args': (7,),  # 保留 7 天
    },
    # 每 30 分钟执行一次
    'update-stats': {
        'task': 'myapp.tasks.update_stats',
        'schedule': 30 * 60,  # 秒
    },
}

任务最佳实践

1. 任务函数参数要简单

python

# ❌ 传递复杂对象，序列化开销大且可能过时
@app.task
def process_user(user):
    user.email = ...
    user.save()

# ✅ 只传 ID，任务内部重新查询
@app.task
def process_user(user_id):
    user = User.objects.get(id=user_id)
    user.email = ...
    user.save()

2. 任务要幂等

任务可能会重复执行（网络问题、重试等），所以要保证多次执行结果一致：

python

@app.task(bind=True, max_retries=3)
def send_sms(self, phone, code):
    try:
        # 发送短信
        sms_client.send(phone, f'验证码：{code}')
    except NetworkError as exc:
        # 网络错误，重试
        raise self.retry(exc=exc, countdown=60)

3. 错误处理和重试

python

@app.task(
    bind=True,
    max_retries=5,
    default_retry_delay=60,  # 默认重试间隔 60 秒
    autoretry_for=(ConnectionError, TimeoutError),  # 自动重试的异常
)
def call_external_api(self, url):
    response = requests.get(url)
    return response.json()

4. 任务状态追踪

python

@app.task(bind=True)
def long_running_task(self, data):
    for i, item in enumerate(data):
        # 更新任务进度
        self.update_state(
            state='PROGRESS',
            meta={'current': i, 'total': len(data)}
        )
        process_item(item)
    return {'status': 'completed'}

适用场景

发送通知：邮件、短信、推送通知
数据处理：批量导入导出、报表生成、图片处理
第三方调用：调用外部 API、支付处理
定时任务：数据清理、统计报表、备份
削峰填谷：突发流量缓冲到队列中慢慢处理
后台计算：机器学习推理、复杂计算

Celery 的不足

监控不完善：需要借助 Flower 等工具
配置复杂：Broker、Backend、队列、路由等配置较多
不支持任务优先级（RabbitMQ 可以通过多个队列实现）
任务结果可能丢失：取决于 Backend 的配置
学习曲线：高级特性（路由、链、组、和弦）较复杂

常见问题

Q：任务执行失败了怎么排查？ A：看 Worker 日志、用 Flower 监控、检查 Backend 中的任务状态和 traceback。

Q：如何保证任务不丢失？ A：用 RabbitMQ（支持消息确认机制）、设置任务持久化、合理配置重试。

Q：Celery 和 Django Q、RQ 比有什么优势？ A：Celery 功能最完善、生态最成熟、支持分布式部署。RQ 更简单轻量。Django Q 是 Django 生态的。

追问延伸：

Celery 的任务路由是什么？怎么把不同任务分到不同队列？
什么是任务链（chain）、组（group）、和弦（chord）？
如何监控 Celery 的运行状态？用过 Flower 吗？
任务执行时间太长怎么处理？（超时、心跳）
如果 Broker 挂了怎么办？怎么保证任务不丢？
你们的 Celery 部署架构是怎样的？有多少个 Worker？
Celery Beat 是单点的吗？怎么实现高可用？

Q1：Django、Flask、FastAPI 三大框架的区别和选型？ 「🟡 中级」 ​

核心定位对比 ​

各框架详解 ​

选型建议 ​

Q2：Django 的 MTV 架构和 MVC 有什么区别？ 「🟢 校招/初级」 ​

MVC 模式 ​

Django 的 MTV 模式 ​

对应关系 ​

为什么叫 MTV 而不叫 MVC？ ​

Django 的完整请求流程 ​

Q3：Django 的请求处理流程？中间件机制？ 「🟡 中级」 ​

Django 请求处理完整流程 ​

中间件机制 ​

中间件的五个钩子方法 ​

执行顺序 ​

常用内置中间件 ​

自定义中间件示例 ​

关键要点 ​

Q4：Django ORM 的原理？select_related 和 prefetch_related 的区别？ 「🟡 中级」 ​

ORM 基本原理 ​

QuerySet 的惰性查询 ​

N+1 查询问题 ​

select_related 和 prefetch_related ​

select_related：JOIN 查询 ​

prefetch_related：两次查询 + Python 关联 ​

对比总结 ​

进阶：Prefetch 对象 ​

其他 ORM 优化技巧 ​

Q5：Django 的信号（Signal）是什么？应用场景？ 「🟡 中级」 ​

信号是什么？ ​

Django 内置信号 ​

信号的使用 ​

方式一：使用 receiver 装饰器 ​

方式二：使用 connect 方法 ​

自定义信号 ​

应用场景 ​

注意事项和坑 ​

信号 vs 直接调用 vs 中间件 ​

Q6：Flask 的上下文机制？请求上下文和应用上下文？ 「🟡 中级」 ​

为什么需要上下文？ ​

两种上下文 ​

1. 请求上下文（Request Context） ​

2. 应用上下文（App Context） ​

上下文栈 ​

为什么需要两种上下文？ ​

LocalProxy 原理 ​

上下文的生命周期 ​

手动使用上下文 ​

g 对象的使用 ​

常见问题 ​

Q7：Flask 的蓝图（Blueprint）是什么？有什么用？ 「🟢 校招/初级」 ​

蓝图是什么？ ​

为什么需要蓝图？ ​

蓝图的使用 ​

创建蓝图 ​

注册蓝图 ​

蓝图的作用 ​

蓝图的模板和静态文件 ​

url_for 与蓝图 ​

蓝图 vs Django 的 app ​

蓝图的局限性 ​

Q8：FastAPI 的特点和原理？为什么性能高？ 「🟡 中级」 ​

FastAPI 是什么？ ​

FastAPI 的核心特点 ​

1. 类型提示驱动（Type Hints Driven） ​

2. 自动生成 API 文档 ​

3. 原生异步支持 ​

4. 高性能 ​

为什么 FastAPI 性能高？ ​

1. 基于 ASGI 而非 WSGI ​

2. 基于 Starlette ​

3. 基于 Pydantic V2 ​

4. 依赖注入系统 ​

FastAPI vs Flask vs Django 性能对比 ​

FastAPI 的适用场景 ​

FastAPI 的不足 ​

Q9：Pydantic 的原理和使用场景？ 「🟡 中级」 ​

Pydantic 是什么？ ​

核心功能 ​

1. 数据模型定义 ​

Q1：Django、Flask、FastAPI 三大框架的区别和选型？「🟡 中级」

核心定位对比

各框架详解

选型建议

Q2：Django 的 MTV 架构和 MVC 有什么区别？「🟢 校招/初级」

MVC 模式

Django 的 MTV 模式

对应关系

为什么叫 MTV 而不叫 MVC？

Django 的完整请求流程

Q3：Django 的请求处理流程？中间件机制？「🟡 中级」

Django 请求处理完整流程

中间件机制

中间件的五个钩子方法

执行顺序

常用内置中间件

自定义中间件示例

关键要点

Q4：Django ORM 的原理？select_related 和 prefetch_related 的区别？「🟡 中级」

ORM 基本原理

QuerySet 的惰性查询

N+1 查询问题

select_related 和 prefetch_related

select_related：JOIN 查询

prefetch_related：两次查询 + Python 关联

对比总结

进阶：Prefetch 对象

其他 ORM 优化技巧

Q5：Django 的信号（Signal）是什么？应用场景？「🟡 中级」

信号是什么？

Django 内置信号

信号的使用

方式一：使用 receiver 装饰器

方式二：使用 connect 方法

自定义信号

应用场景

注意事项和坑

信号 vs 直接调用 vs 中间件

Q6：Flask 的上下文机制？请求上下文和应用上下文？「🟡 中级」

为什么需要上下文？

两种上下文

1. 请求上下文（Request Context）

2. 应用上下文（App Context）

上下文栈

为什么需要两种上下文？

LocalProxy 原理

上下文的生命周期

手动使用上下文

g 对象的使用

常见问题

Q7：Flask 的蓝图（Blueprint）是什么？有什么用？「🟢 校招/初级」

蓝图是什么？

为什么需要蓝图？

蓝图的使用

创建蓝图

注册蓝图

蓝图的作用

蓝图的模板和静态文件

url_for 与蓝图

蓝图 vs Django 的 app

蓝图的局限性

Q8：FastAPI 的特点和原理？为什么性能高？「🟡 中级」

FastAPI 是什么？

FastAPI 的核心特点

1. 类型提示驱动（Type Hints Driven）

2. 自动生成 API 文档

3. 原生异步支持

4. 高性能

为什么 FastAPI 性能高？

1. 基于 ASGI 而非 WSGI

2. 基于 Starlette

3. 基于 Pydantic V2

4. 依赖注入系统

FastAPI vs Flask vs Django 性能对比

FastAPI 的适用场景

FastAPI 的不足

Q9：Pydantic 的原理和使用场景？「🟡 中级」

Pydantic 是什么？

核心功能

1. 数据模型定义