Python Web 进阶 · Django 深入 / Celery / 缓存 / DRF / 部署

本页定位 · Python Web 进阶

《Python Web 开发》带你用 Flask / FastAPI / Django 写出了能跑的接口。本页进入真实项目里绕不开的工程问题：Django ORM 的高级查询与索引、异步任务 Celery 的可靠投递、缓存与性能、信号与节流、DRF 把接口规范化、以及生产部署。我会像讲《Java 进阶》那样，把每个点拆成"它是什么 → 为什么这样设计 → 怎么用 → 踩过什么坑"。基线：Django 5、Celery 5、Django REST Framework 3.15、Gunicorn 21、uvicorn 0.30。

Django ORM 深入：查询优化、N+1、索引

QuerySet · select_related · prefetch_related · Index

会写 Model.objects.filter() 只是起点。QuerySet 是惰性求值的，懂得"什么时候真正打库、怎么减少 SQL 数量"，是 Django 性能的第一道关。这一章把"对象关系映射背后发生了什么"讲透。

惰性求值：查询什么时候才真正打到数据库

很多人以为写完 qs = Book.objects.all() 数据库就被查了——其实没有。QuerySet 只在真正需要结果时才执行：切片、迭代、len()、list()、bool()、或访问字段。理解这点，你才明白"为什么两个 QuerySet 多次求值会查两次库"。

# 下面这几行，数据库一次都没碰
qs = Book.objects.filter(pub__year__gt=2020)
# 只有在这里（求值时）才真正发 SQL
for b in qs:          # 第 1 次求值 → 1 条 SQL
    print(b.title)
len(qs)                # 第 2 次求值 → 又发 1 条 SQL（缓存的是结果集，不是查询计划）

论惰性既是优点也是坑

优点是可链式组合、可复用、可延迟到真正需要时。坑在于在循环里反复触发求值，或在模板里多次迭代同一 QuerySet，把一次查询变成 N 次。把求值集中到一次（list(qs) 或 prefetch 后迭代），是 Django 性能第一招。

N+1 问题：列表页最常见的性能杀手

场景：列出 100 本书，每本显示作者名字。朴素写法是查 100 本书（1 条 SQL），再对每本书访问 book.author.name（各自查作者的 100 条 SQL）——共 101 条，这就是 N+1。列表页一卡，八成是它。

# 反例：N+1（100 本书 → 1 + 100 条 SQL）
books = Book.objects.all()[:100]
for b in books:
    print(b.author.name)   # 每本都触发一次 author 查询

# 正解：select_related 用 SQL JOIN 一次取出（外键/一对一）
books = Book.objects.select_related('author').all()[:100]

# 多对多 / 反向关系用 prefetch_related（独立查询后内存关联）
books = Book.objects.prefetch_related('tags', 'author').all()[:100]

方法	适用关系	底层做法	何时用
select_related	外键、一对一	SQL JOIN 一次取主表+从表	"每本书都要作者名"这类强关联
prefetch_related	多对多、反向 FK	先查主表，再 IN 查从表，内存里拼	"每本书的标签列表"这类集合

prefetch_related 救不了"带过滤的关联"

如果你想"每本书只取它的'已发布'标签"，prefetch_related('tags') 会取出全部标签再过滤，浪费。要用 Prefetch('tags', queryset=Tag.objects.filter(active=True)) 指定子查询，精准控制预取内容。

聚合、注解与 F 表达式：把计算下推到数据库

能不在 Python 里循环求和，就别循环。Django 提供 aggregate（汇总成字典）、annotate（给每行加派生字段）、以及 F 表达式（引用数据库字段做原子运算）。

from django.db.models import Count, Avg, Sum, F

# 每个分类下有多少本书（按分组注解）
cats = Category.objects.annotate(num_books=Count('book'))

# 平均评分（聚合，返回 dict）
stat = Book.objects.aggregate(avg=Avg('rating'), total=Sum('sales'))

# F 表达式：库存 - 已售，原子更新，避免"读-改-写"竞态
Book.objects.filter(id=1).update(stock=F('stock') - F('sold'))

论为什么用 F 而不是直接算

stock = book.stock - book.sold; book.save() 是"先读再写"，两个请求并发会丢更新（都读到旧值）。F 让运算发生在数据库一侧，是原子的，且少一次网络往返。凡是"基于自身字段更新"，优先 F。

索引：让查询从全表扫描变成定向查找

没有索引，filter(email=...) 要扫全表。给常被查询/排序/连表的字段加索引，是性价比最高的优化。Django 在模型字段上加 db_index=True，或建复合索引 indexes。

class Book(models.Model):
    title = models.CharField(max_length=200)
    author = models.ForeignKey('Author', on_delete=models.CASCADE, db_index=True)
    pub_date = models.DateField()

    class Meta:
        # 复合索引：按作者+出版年查询时命中
        indexes = [models.Index(fields=['author', '-pub_date'], name='idx_author_date')]
        # 唯一索引：邮箱不能重复
        constraints = [models.UniqueConstraint(fields=['isbn'], name='uniq_isbn')]

索引不是越多越好

索引会拖慢写入（每次 INSERT/UPDATE 要维护索引），还占空间。只对高频查询、高选择性的字段加索引；像"性别"这种只有两三个值的字段，加索引几乎没用（选择性太低，优化器往往直接全表扫）。改完索引记得 makemigrations + migrate。

事务与批量：一致性、并发与吞吐

"下单一本书要减库存、加订单、记日志"——这三步必须要么全成要么全败，用事务包起来。高并发扣库存用 select_for_update 加行锁，避免超卖。

from django.db import transaction

with transaction.atomic():           # 一个事务
    book = Book.objects.select_for_update().get(id=1)  # 锁住行，别人同时改会等
    if book.stock <= 0:
        raise ValueError("售罄")
    book.stock -= 1
    book.save()

# 批量插入：一次 SQL 插 1000 行，比循环 save() 快几十倍
Book.objects.bulk_create([Book(title=f"b{i}") for i in range(1000)], batch_size=500)

诊断工具：别靠猜，看真实 SQL

怀疑慢查询时，先用 django-debug-toolbar 看每个请求发了多少条 SQL、各耗时多少；上线后用 QuerySet.explain() 看数据库的执行计划，确认有没有走索引。

# 在 shell 里直接看执行计划（PostgreSQL 示例）
print(Book.objects.filter(author_id=3).order_by('-pub_date').explain(verbose=True))
# 输出会告诉你：Index Scan using idx_author_date ... 还是 Seq Scan（全表扫，要警惕）

# 临时打开所有 SQL 日志（开发期）
import logging
logging.getLogger('django.db.backends').setLevel(logging.DEBUG)

论优化顺序：先量再改

现象（页面慢）→ 用 debug-toolbar / EXPLAIN 定位慢点（N+1？全表扫？）→ 选手段（select/prefetch、加索引、批量）→ 改完再量一遍确认下降。没度量就开始"优化"等于闭眼开车。

只查需要的字段：only / defer / values

ORM 默认 SELECT *，把整行所有字段都取回来。但列表页往往只要 id 和 title，多余字段（大文本、JSON 字段）白白占带宽和内存。用 only() 只取指定列、defer() 延迟大字段、values() 直接拿 dict 不要对象。

方法	效果	何时用
only('title')	只查列出的字段，其余访问时懒加载	列表页少用字段
defer('body')	除列出的大字段外都查，大字段用时再取	跳过大文本 / JSON 列
values('id','title')	返回 dict 而非 Model 对象	纯展示、不要对象行为

Celery 异步任务：broker/worker/重试/链路追踪

Celery · Broker · Worker · Beat · Idempotent

发邮件、生成报表、调第三方——这些不该阻塞 HTTP 请求。Celery 把它们丢到后台 Worker 异步执行，让 Web 进程秒回响应。但"异步"带来的是可靠性问题：任务丢了怎么办、重复了怎么办、卡住了怎么查。

四个角色：任务从产生到执行经过谁

Celery 不是单一进程，而是一套协作。搞清每个角色，排查时有方向。

角色	作用	常见选型
Producer	你的 Web 代码，调用 `.delay()` 产生任务	Django 视图
Broker	消息队列，暂存任务消息	Redis / RabbitMQ
Worker	真正执行任务的进程	celery worker 进程
Backend	存任务结果/状态（可选）	Redis / 数据库
Beat	定时把任务投进 Broker	celery beat

定义任务：绑定 self、重试、超时

任务函数用 @app.task 装饰。bind=True 让函数第一个参数是任务自身（self），从而能调用 self.retry()。临时故障（网络抖动）要自动重试，且重试要带退避。

# tasks.py —— 可重试的邮件任务
@app.task(bind=True, max_retries=3, default_retry_delay=30)
def send_welcome_email(self, user_id):
    try:
        do_send(user_id)
    except TemporaryError as e:
        # 指数退避：第 n 次重试等 30 * 2^(n-1) 秒
        raise self.retry(exc=e, countdown=30 * (2 ** self.request.retries))

论异步任务必须幂等

任务可能因重试、Worker 崩溃重发、broker 重复投递而执行多次。写任务时假设"同一任务会跑两次"——用唯一键去重、用状态机防重入，否则重试会变成"发两次邮件、扣两次钱"。幂等是 Celery 工程的第一铁律。

定时任务：Beat 让作业自己跑

每天汇总、每小时清理过期会话——这类周期性工作交给 celery beat。它按时间表把任务投进 Broker，由 Worker 执行。配置用 crontab 或 schedule（类似 timedelta）。

# celery_app.py —— 周期性任务表
from celery.schedules import crontab

app.conf.beat_schedule = {
    'summarize-every-night': {
        'task': 'tasks.daily_summary',
        'schedule': crontab(hour=2, minute=0),   # 每天 02:00
    },
    'ping-every-5min': {
        'task': 'tasks.health_ping',
        'schedule': 300.0,                     # 每 300 秒
    },
}

Beat 不要多开实例

如果跑两个 beat 进程，同一个定时任务会被投递两次。生产里 beat 只起一个（配合锁或单副本部署）。周期任务"重复执行"也要求它幂等。

链路追踪：把一次请求和它的后台任务串起来

用户下单（Web 请求）触发了 3 个异步任务，其中一个失败了——你怎么知道"是这次下单导致的"？答案是把 Web 请求的 traceId 透传给每个任务，在日志里统一带 traceId，事后按 traceId 串联全链路。

# 把当前请求的 trace_id 作为任务参数透传
trace_id = get_current_trace()
generate_report.delay(order_id, trace_id)

@app.task(bind=True)
def generate_report(self, order_id, trace_id):
    log("start", trace_id=trace_id)   # 每行日志都带 trace_id
    ...

论为什么不直接传 ORM 对象

新手常写 my_task.delay(user) 想把对象传进去——不行。任务参数必须可序列化（JSON/pickle），ORM 对象带数据库连接、不可序列化。正确做法是只传主键 user_id，在任务里重新 User.objects.get(id=user_id) 查。这样既安全又避免传过期数据。

监控与排错：任务卡住、堆积怎么看

flower 是 Celery 的实时仪表盘：看队列深度、Worker 心跳、任务成功率，及时发现积压。关键纪律是"失败不能静默消失"——给任务配 on_failure 钩子，把异常发到告警群或写库；配合带 task_id、trace_id 的结构化任务日志，出问题能按 task_id 反查现场。

论broker 健康检查常被忽略

Worker 与 Redis / RabbitMQ 之间也有连接池。broker 抖动时 Worker 会重连，但那段时间产生的任务可能短暂失败——这类"可重试"故障靠前面讲的 self.retry() 兜底。生产一定单独监控 broker 自身（内存、堆积长度），broker 挂了，所有异步能力一起没。

Worker 配置与优雅退出

Worker 不是 celery worker 一把梭就行。并发模型（prefork / gevent）、并发数、预取数（prefetch_multiplier）都影响吞吐与内存。滚动发布时要先让 Worker 处理完手头任务再杀（graceful shutdown）。

# prefork 模式，4 个进程，单任务预取 4 条
celery -A proj worker -c 4 --prefetch-multiplier=4
# 发 SIGTERM 后，Worker 会先跑完 in-flight 任务再退出（默认等待 soft/hard timeout）

prefork + 大内存对象 = OOM

prefork 模式每个子进程都复制一份内存。如果你的任务会加载大模型 / 大词典到内存，多进程会让内存翻 N 倍。这种场景要么用 --concurrency=1，要么换 gevent 协程模式，要么把大对象放到共享存储按需取。

缓存（Redis 层/会话）、信号、节流

Cache Framework · Redis · Signal · Throttle

扛读流量、降数据库压力，缓存是第一杠杆。Django 自带多级缓存框架，Redis 是最常用的后端。这一章还讲信号（解耦横切逻辑）和节流（限制滥用）。

多级缓存：从内存到 Redis 到 CDN

Django 缓存框架抽象了一层，后端可换：开发用 LocMemCache（进程内，最快但不跨进程），生产用 RedisCache（跨进程、可持久）。越靠近用户越快。

# settings.py 指定 Redis 为默认缓存后端
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.redis.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
        'TIMEOUT': 300,                 # 默认 5 分钟过期
    }
}

视图级与低级 API：两种用法

不想动业务代码时，@cache_page 把整个响应缓存；要精细控制时，用 cache.set/get 低级 API 缓存计算结果。

# 视图级缓存：命中则直接返回，不进视图函数
from django.views.decorators.cache import cache_page

@cache_page(60 * 15)
def hot_list(request):
    return JsonResponse(compute_expensive_list())

# 低级 API：缓存一个昂贵计算结果
from django.core.cache import cache
data = cache.get('stats')
if data is None:
    data = compute_stats()          # 缓存未命中才算
    cache.set('stats', data, timeout=900)
return data

论用 cache 而不是全局 dict

新手为了"快"把数据塞进 Python 全局变量——不行：多进程 Worker 各自一份、重启即丢、无法共享。缓存必须放外部存储（Redis），所有进程共享、可过期、可跨重启。

Cache-Aside 与失效：三难题里最难的是失效

"缓存三难题"指缓存穿透、击穿、雪崩，但工程上最磨人的是失效：数据变了，缓存还是旧的，用户看到陈旧内容。推荐的 Cache-Aside 模式：读时查缓存、没有再查库并回填；写时更新数据库后主动删缓存（下次读自动重建）。

# 写操作后主动失效相关缓存
def update_profile(user_id, data):
    Profile.objects.filter(id=user_id).update(**data)
    cache.delete(f'profile:{user_id}')   # 删，而不是存旧值

缓存三大灾难

穿透：查不存在的 key，每次都打库 → 缓存空值（短 TTL）或用布隆过滤器。击穿：热点 key 过期瞬间海量请求打库 → 互斥锁重建或不过期+后台刷新。雪崩：大量 key 同一时刻过期 → TTL 加随机抖动，避免同时失效。

会话放进 Redis：多进程共享登录态

用 Django 默认（数据库/文件）会话，多 Worker 还行；但上了多机部署或想减轻数据库压力，把 session 存 Redis 更稳。配置上把 SESSION_ENGINE 设为 django.contrib.sessions.backends.cache 并指向 Redis 缓存别名即可；登录态跨进程、跨重启都还在，扩容 Web 时用户不掉线。

信号：把横切逻辑从主流程里摘出来

post_save、pre_delete 等信号能在"模型保存后"自动触发逻辑（比如用户注册后发欢迎邮件）。它解耦了"主业务"和"顺带做的事"。

from django.db.models.signals import post_save
from django.dispatch import receiver

@receiver(post_save, sender=User)
def on_user_created(sender, instance, created, **kwargs):
    if created:                        # 仅新建时
        send_welcome_email.delay(instance.id)

信号容易被滥用

信号藏在别处、执行时机隐式，难调试、易产生意外副作用（循环触发、测试时偷偷发邮件）。能用显式调用或事件总线表达清楚的，优先显式；信号只留给"确实与主业务正交的横切逻辑"（审计日志、通知）。

节流：别让一个用户打爆你的接口

无论缓存多强，恶意刷接口都会拖垮系统。节流（rate limiting）限制单位时间内的请求数。简单场景可用 Django 中间件或 Redis 计数；规范接口建议放进 DRF（见第 4 章），这里先给一个手写的 Redis 滑动窗口思路。

# 基于 Redis 的简单计数限流（每用户每分钟 60 次）
def is_allowed(user_id, limit=60, window=60):
    key = f'rl:{user_id}'
    n = cache.incr(key)          # 原子自增
    if n == 1:
        cache.expire(key, window)
    return n <= limit

论限流放在哪一层

边缘（Nginx / API 网关）挡大流量、最便宜；应用层（DRF throttle）做精细到用户/接口的限流；两者互补。别只靠应用层——恶意流量早把你的进程占满了才轮到限流逻辑。

Django REST Framework：序列化/分页/权限/限流/版本化

DRF · Serializer · ViewSet · Pagination · Throttle

手写 API 容易各写各的、字段规则重复、安全漏配。DRF 提供序列化、视图集、权限、分页、限流一套规范，前后端协作更顺，也把安全点集中管起来。

序列化器：统一的校验/转换/写出

Serializer 把"模型对象 ↔ JSON"的边界管好：定义字段、校验输入、控制输出（哪些字段只读/只写/嵌套）。比手写 dict 拼装更安全、可复用。

class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['id', 'name', 'email']
        extra_kwargs = {'email': {'write_only': True}}  # 写入接收、输出不回显（防泄露）

    # 字段级校验：用户名不能太短
    def validate_name(self, value):
        if len(value) < 2:
            raise serializers.ValidationError("名字至少 2 个字符")
        return value

论为什么用序列化器而非手动拼 dict

Serializer 把校验、转换、写出三件事统一了，避免每个接口重复写字段规则；配合 ViewSet 路由自动生成，减少样板；权限/限流集中声明，安全不漏。是 DRF 的"地基"。

视图集与路由：一套 CRUD 自动生成

ModelViewSet 把 list / create / retrieve / update / destroy 全包了，配合 DefaultRouter 自动生成 RESTful 路由，不用手写一堆 path()。

class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer
    permission_classes = [IsAuthenticated]

# urls.py
router = DefaultRouter()
router.register(r'users', UserViewSet)
urlpatterns = router.urls   # 自动得到 users/  users/{pk}/ 等

分页：别一次吐 10 万条

不分页的列表接口，数据一多就拖垮数据库和带宽。DRF 提供三种分页：按页码、按偏移、按游标（游标分页对大数据集最稳，且不受中间插入影响）。

分页类	URL 形态	适用
PageNumberPagination	`?page=2&page_size=20`	普通后台列表
LimitOffsetPagination	`?limit=20&offset=40`	无限滚动前端
CursorPagination	`?cursor=xxx`	大数据集、实时流

REST_FRAMEWORK = {
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
    'PAGE_SIZE': 20,
}

权限与认证：谁能看、谁能动

权限分两层：认证（你是谁，拿到身份）和授权（你能干什么）。DRF 提供 IsAuthenticated、IsAdminUser、以及可自定义的权限类。对象级权限（"只能改自己的文章"）要重写 has_object_permission。

class IsOwner(BasePermission):
    def has_object_permission(self, request, view, obj):
        return obj.author == request.user   # 只有作者本人能改

class PostViewSet(viewsets.ModelViewSet):
    permission_classes = [IsAuthenticated, IsOwner]

限流：接口级别掐住滥用

比第 3 章的全局节流更精细——DRF 的 throttle 可限制到"每个用户 / 每个 IP / 每个视图"的速率，且能区分"已登录"和"匿名"两套配额。

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': ['rest_framework.throttling.AnonRateThrottle',
                                   'rest_framework.throttling.UserRateThrottle'],
    'DEFAULT_THROTTLE_RATES': {'anon': '10/min', 'user': '100/min'},
}

class SmsViewSet(viewsets.ViewSet):
    throttle_classes = [UserRateThrottle]   # 发短信接口更要限流，防刷

限流配置忘了加全局才不生效

只在某个视图写 throttle_classes 只对该视图生效。要全站默认限流，必须在 REST_FRAMEWORK['DEFAULT_THROTTLE_CLASSES'] 里配置，否则大多数接口还是裸奔。

版本化：API 变了，老用户别崩

接口难免演进。版本化让你"旧版本继续服务，新版本并行发布"，给老客户端留出迁移窗口。常见三种做法：

方式	形态	优点 / 缺点
URL 路径	`/api/v1/users/`	直观、易调试；URL 冗长
Accept 头	`Accept: application/vnd.myapi.v2+json`	URL 干净；不直观、难手测
命名空间	不同 urls 模块挂载 v1 / v2	代码隔离清晰；维护两份

# URL 路径版本化：最简单也最常用
urlpatterns = [
    path('api/v1/', include('api.v1.urls')),
    path('api/v2/', include('api.v2.urls')),
]

部署：gunicorn/uvicorn、容器、静态文件、WSGI/ASGI

Gunicorn · uvicorn · Docker · Nginx · WSGI/ASGI

runserver 只用于开发。生产要用应用服务器 + 反向代理，还要处理静态文件、环境变量、迁移、HTTPS。这一章把"从能跑到能扛"补齐。

WSGI 与 ASGI：同步与异步的两条路

Django 传统是 WSGI（同步，每请求占一个 worker 线程/进程）。若用了异步视图、Channels（WebSocket），需要 ASGI。选哪个取决于你是否真的有异步需求。

协议	服务器	适合
WSGI	Gunicorn / uWSGI	纯同步 Django 视图
ASGI	uvicorn / daphne / hypercorn	异步视图、WebSocket、长连接

Gunicorn：同步部署的主力

Gunicorn 是 WSGI 服务器，常用多进程模型扛并发。Worker 数经验公式 2 * CPU核数 + 1，太多会争抢内存和 GIL。

# 同步（WSGI）：4 个 worker
gunicorn project.wsgi:application -w 4 -b 0.0.0.0:8000

# 用 gevent 协程模式处理较多 IO 等待（适合 IO 密集）
gunicorn project.wsgi:application -k gevent -w 2 --worker-connections 1000

论为什么 Worker 数不是越多越好

Python 有 GIL，单进程内多线程并不能真正并行 CPU 任务。多 worker = 多进程，能利用多核，但每个进程吃一份内存（含 Django、模型、缓存客户端）。超过 2*CPU+1 后，进程间争抢内存带宽和上下文切换成本反而拉低吞吐。

uvicorn 与 ASGI：异步部署

当你的视图用 async def、或用了 Channels，Gunicorn 的同步 worker 跑不了，要上 uvicorn（ASGI）。生产通常 gunicorn + uvicorn worker 组合，既有多进程管理又有异步能力。

# 纯 uvicorn（简单场景）
uvicorn project.asgi:application -w 4

# 生产推荐：Gunicorn 管进程 + UvicornWorker 跑异步
gunicorn project.asgi:application -k uvicorn.workers.UvicornWorker -w 4

异步视图里别调同步阻塞

在 async def 视图里直接调 time.sleep、requests.get 这类同步阻塞，会卡住整个事件循环，所有并发请求一起等。异步视图里 IO 要用 aiohttp / httpx.AsyncClient，或把阻塞活丢给 sync_to_async / 线程池。

静态文件：collectstatic 交出去

Django 开发时自动serve静态文件，生产不行。要把所有 app 的静态资源收集到统一目录（collectstatic），由 Nginx 或 CDN 直接托管——既快又减轻 Python 进程负担。

# settings.py
STATIC_URL = '/static/'
STATIC_ROOT = '/var/www/static/'     # collectstatic 输出目录

# 收集命令（部署时跑一次）
python manage.py collectstatic --noinput

# Nginx 把 /static/ 直接指到该目录，不走 Django

Docker 化：环境一致、随处可跑

把应用、依赖、运行方式打进镜像，开发/测试/生产环境一致，"我本地是好的"从此消失。多进程（web + worker + beat）用 docker-compose 或 K8s 编排。

# Dockerfile（多阶段，精简镜像）
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
RUN python manage.py collectstatic --noinput
EXPOSE 8000
CMD ["gunicorn", "project.wsgi:application", "-w", "4", "-b", "0.0.0.0:8000"]

论部署清单（上线前逐项勾）

① 静态文件：collectstatic 交 Nginx/CDN。② 环境变量：SECRET_KEY、数据库密码走环境变量（django-environ），绝不进代码。③ 迁移：CI/CD 里跑 migrate。④ Worker 数：按 CPU/内存定，别开爆。⑤ 日志：结构化、可检索。⑥ HTTPS：Nginx 终止 + HSTS。⑦ 启动命令：用 Gunicorn 而非 runserver。

配置管理：12-Factor 与 django-environ

把配置从代码里剥离，靠环境变量注入，是 12-Factor 应用的核心。用 django-environ 从 .env 读，本地开发与生产共用同一套代码、不同配置。

import environ
env = environ.Env(DEBUG=(bool, False))
environ.Env.read_env()                 # 读 .env

DEBUG = env('DEBUG')
DATABASES = {'default': env.db('DATABASE_URL')}   # 一行解析 DATABASE_URL
SECRET_KEY = env('SECRET_KEY')

别把 SECRET_KEY / 密码提交进 Git

泄漏 SECRET_KEY 等于把会话签名、密码重置令牌的钥匙交出去，攻击者可伪造会话。.env 必须进 .gitignore，生产用密钥管理服务（Vault / 云 Secrets）注入。CI 里扫描一下有没有敏感串。

编排：web + worker + beat 一起起来

真实部署不是只跑一个 web 进程，而是一组：web（接请求）、worker（跑异步）、beat（投定时）、可能还有 channels 层。用 docker-compose 或 K8s Deployment 把它们在同一个应用下编排起来，各自独立扩缩容。

进程	命令	扩缩容依据
web	`gunicorn ... -w 4`	HTTP QPS
worker	`celery -A proj worker -c 4`	队列堆积长度
beat	`celery -A proj beat`	单副本即可

论滚动发布时别让任务丢

更新 worker 镜像时，先给旧 worker 发 SIGTERM，它处理完手头的任务再退出（graceful）；同时新 worker 已起，流量无缝。若直接 SIGKILL，正在跑的任务会丢失、且可能重复（被 broker 重新投递）——这里任务幂等又救了你一次。

类型注解 / 可观测 / 安全

typing · mypy · logging · metrics · security

能跑只是起点，可维护、可观测、可被攻击面可控才是生产级。这一章讲 Python Web 的"工程化三件套"：类型注解、可观测性、安全底线。

类型注解与 mypy：给动态语言加护栏

Python 是动态类型，大项目里"传错类型"要等到运行时才炸。类型注解（typing）是可选但强烈建议的文档+检查：配合 mypy 在提交前静态找出类型错误，比线上报错便宜得多。

from typing import Optional, List, Dict
from django.http import HttpRequest

def get_user_orders(uid: int, status: Optional[str] = None) -> List[Dict[str, int]]:
    # 入参、返回值都标了类型
    ...

# 运行 mypy 检查（CI 里卡住类型错误）
# mypy project/  → 报告 "incompatible type" 等

论为什么 Django 项目也值得上 mypy

Django 的"魔法"（动态属性、管理器）让 IDE 和静态检查很难——但 mypy 配合 django-stubs 插件能理解 ORM。收益：重构时编辑器立刻标红错误处、新人读代码有类型线索、减少"传了个 None 进来崩了"的线上事故。渐进式开启即可，不必一步到位。

日志：结构化、带 trace、别打敏感

用标准 logging 模块，配置 JSON 格式输出（方便 ELK / Loki 采集）。每条请求带 trace_id，把分散在多行的日志串成一条链路。

import logging, json
log = logging.getLogger(__name__)

# 结构化日志：用 dict 传递字段，下游可检索
log.info("order_placed", extra={'trace_id': tid, 'order_id': oid, 'amount': amt})

# 千万别这么干：把密码/身份证打进日志
log.info(f"login {user} pwd={password}")   # 合规事故

日志的三个雷

① 打敏感信息（密码、token、身份证）进日志，是合规红线。② 循环里打 DEBUG，日志量爆炸写满磁盘。③ 用 f-string 拼好后不打印也先格式化，高开销。用惰性参数（extra= / 占位）或调低级别控制。

Metrics：用数字看见系统健康

日志看"发生了什么"，指标看"系统多健康"。用 django-prometheus 暴露 QPS、耗时、错误率，喂给 Prometheus，配 Grafana 面板与告警。

# settings.py 接入 prometheus 中间件
INSTALLED_APPS += ['django_prometheus']
MIDDLEWARE = ['django_prometheus.middleware.PrometheusBeforeMiddleware'] + MIDDLEWARE
MIDDLEWARE += ['django_prometheus.middleware.PrometheusAfterMiddleware']

# 业务指标：自定义计数器
from prometheus_client import Counter
ORDERS = Counter('orders_total', '订单总数', ['status'])

分布式追踪：一次请求穿过多少服务

一个用户请求可能串起 Web → Celery → 数据库 → 第三方 API。OpenTelemetry 给每次请求发一个 traceId，自动记录各段耗时，定位"到底慢在哪一段"。

# 用 opentelemetry-instrumentation-django 自动埋点
# 启动前加环境变量即可，无需改业务代码：
OTEL_SERVICE_NAME=web OTEL_EXPORTER_OTLP_ENDPOINT=http://collector:4317 \
  opentelemetry-instrument gunicorn project.wsgi:application -w 4

论Metrics / 日志 / 追踪各管一摊

Metrics：QPS、耗时分位、错误率，配告警，回答"系统健康吗"。日志：结构化、集中采集，回答"具体发生了什么"。追踪：一次跨服务请求全链路耗时，回答"慢在哪"。三者互补，缺一个排障就少一只眼。

安全底线：CSRF / XSS / SQL 注入 / 依赖

Web 安全的经典三板斧：CSRF（借浏览器 cookie 伪造请求）、XSS（注入脚本偷信息）、SQL 注入（拼接 SQL）。Django 用 ORM、模板自动转义、CSRF 中间件帮了大忙，但用错方式会破防。

# ① 永远用 ORM 参数化，别手拼 SQL（避免注入）
User.objects.filter(name=request.GET['q'])          # 安全
# User.objects.raw("SELECT * FROM user WHERE name='" + q + "'")  # 致命

# ② 关闭 CSRF 保护的接口务必自己校验来源
from django.views.decorators.csrf import csrf_exempt
@csrf_exempt
def webhook(request):
    if request.headers.get('X-Signature') != expected:
        raise PermissionDenied

csrf_exempt 不是免死金牌

第三方回调（支付、Webhook）通常带不了 CSRF token，于是开发者随手 @csrf_exempt。但这等于放弃了防伪造保护——必须用签名校验 / IP 白名单 / HMAC 替代。另外模板里用户输入一定要走 {{ value }} 自动转义，别用 |safe 裸渲染用户内容，否则 XSS 漏洞。

依赖安全：你的第三方库就是攻击面

项目 90% 的代码来自依赖。一个被投毒的包、一个未修的 CVE，就能让你整站沦陷。两条纪律：① 用 pip-audit -r requirements.txt 扫描已知漏洞（基于 PyPI 漏洞库），把报告接进 CI，有 CVE 就红；② 锁定版本，pip freeze > requirements.lock，避免"明天自动装了个有问题的新版"在生产悄悄生效。

论安全是持续动作不是一次性

上线前扫一遍只是开始。把 pip-audit 接进 CI，依赖有 CVE 就红；订阅 Django 安全公告，及时升小版本（Django 的 x.y.z 里 z 是安全补丁，升它基本零风险）。DEBUG 在生产务必 False，否则错误页会泄露配置与源码。

上线开关：DEBUG 与 ALLOWED_HOSTS

两个最常被"上线后忘记改"的开关。DEBUG=True 在生产会暴露整个堆栈、配置甚至源码，且性能差，必须为 False。ALLOWED_HOSTS 未配置时 Django 拒绝未知 Host 的请求，是防 Host 头攻击的底线。

设置	开发	生产	不配的后果
DEBUG	True（看报错页）	False	泄露配置 / 源码
ALLOWED_HOSTS	[] 可本地	['your.domain']	Host 头攻击、400 拒绝

结

本页要点

Django ORM 靠惰性求值、select/prefetch_related 消 N+1、F 表达式原子更新、索引与事务做性能与一致性；Celery 用 Broker/Worker/Beat 把重活异步化，任务必幂等、只传主键、配重试与追踪；缓存框架（Redis 后端）扛读，Cache-Aside + 防穿透/击穿/雪崩，会话可入 Redis，信号解耦横切逻辑，节流防刷；DRF 用序列化器、ViewSet、分页、权限、限流、版本化把接口规范化；生产用 Gunicorn/uvicorn + Nginx，collectstatic、容器化、配置外置（django-environ）；工程化上 mypy 加类型护栏，结构化日志 + Metrics + 追踪三件套可观测，安全守住 CSRF/XSS/SQL 注入/依赖漏洞底线。

自测 · 看你是否真懂

1.Django 里 select_related 和 prefetch_related 分别在什么场景用？为什么能消 N+1？

查看答案

select_related 用于外键/一对一，用 SQL JOIN 一次取出，避免逐条查；prefetch_related 用于多对多/反向，先查主表再单独查关联表在内存里拼。两者都把"1 + N 条 SQL"压成 1~2 条，消除 N+1。

2.为什么 Celery 任务要设计成"只传主键、不传 ORM 对象"，且必须幂等？

查看答案

任务参数必须可序列化，ORM 对象带连接、不可序列化，所以只传 user_id 并在任务里重查。任务可能因重试、Worker 崩溃重发而重复执行，幂等（唯一键去重、状态机防重入）保证重复执行不产生重复副作用（如重复扣款）。

3.缓存"三难题"指什么？Cache-Aside 怎么解决失效？

查看答案

穿透（查不存在的 key 每次打库）、击穿（热点 key 过期瞬间打库）、雪崩（大量 key 同时过期）。Cache-Aside：读时查缓存、未命中查库并回填；写时更新库后主动删缓存，下次读自动重建，避免用户看到陈旧数据。

4.生产为什么不用 runserver？Gunicorn 的 worker 数一般怎么定？

查看答案

runserver 是单线程开发服务器，无安全防护、无多进程、性能差。生产用 Gunicorn/uvicorn 多 worker + Nginx 反代与静态托管。Worker 数经验公式 2*CPU核数+1，过多会因 GIL 争抢和内存翻倍反而降吞吐。

5.异步视图（async def）里直接调 requests.get 或 time.sleep 有什么后果？

查看答案

会卡住整个事件循环，所有并发请求一起等，异步优势尽失。异步视图里 IO 要用 aiohttp/httpx.AsyncClient，阻塞活丢给 sync_to_async 或线程池；部署层对应走 ASGI（uvicorn）。

下一步往哪走

路学完 Python Web 进阶之后

① 看数据库进阶页：Celery 的 broker、缓存后端、Django 的连接池都连着 Redis / MySQL 这套底座，索引与事务在那里讲得更深。

② 看 DevOps 页：Docker、CI、可观测把部署这一章补全——你写的 Dockerfile 怎么进流水线、日志怎么进 Loki。

③ 看系统设计页：缓存、限流、异步任务正是高可用架构的零件，去那里看它们怎么组合。

④ 动手：搭一个 Django + DRF + Celery + Redis + Docker 的可部署项目，把本页每个组件都接上。

Django ORM 深入：查询优化、N+1、索引

惰性求值：查询什么时候才真正打到数据库

论惰性既是优点也是坑

N+1 问题：列表页最常见的性能杀手

聚合、注解与 F 表达式：把计算下推到数据库

论为什么用 F 而不是直接算

索引：让查询从全表扫描变成定向查找

事务与批量：一致性、并发与吞吐

诊断工具：别靠猜，看真实 SQL

论优化顺序：先量再改

只查需要的字段：only / defer / values

Celery 异步任务：broker/worker/重试/链路追踪

四个角色：任务从产生到执行经过谁

定义任务：绑定 self、重试、超时

论异步任务必须幂等

定时任务：Beat 让作业自己跑

链路追踪：把一次请求和它的后台任务串起来

论为什么不直接传 ORM 对象

监控与排错：任务卡住、堆积怎么看

论broker 健康检查常被忽略

Worker 配置与优雅退出

缓存（Redis 层/会话）、信号、节流

多级缓存：从内存到 Redis 到 CDN

视图级与低级 API：两种用法

论用 cache 而不是全局 dict

Cache-Aside 与失效：三难题里最难的是失效

会话放进 Redis：多进程共享登录态

信号：把横切逻辑从主流程里摘出来

节流：别让一个用户打爆你的接口

论限流放在哪一层

Django REST Framework：序列化/分页/权限/限流/版本化

序列化器：统一的校验/转换/写出

论为什么用序列化器而非手动拼 dict

视图集与路由：一套 CRUD 自动生成

分页：别一次吐 10 万条

权限与认证：谁能看、谁能动

限流：接口级别掐住滥用

版本化：API 变了，老用户别崩

部署：gunicorn/uvicorn、容器、静态文件、WSGI/ASGI

WSGI 与 ASGI：同步与异步的两条路

Gunicorn：同步部署的主力

论为什么 Worker 数不是越多越好

uvicorn 与 ASGI：异步部署

静态文件：collectstatic 交出去

Docker 化：环境一致、随处可跑

论部署清单（上线前逐项勾）

配置管理：12-Factor 与 django-environ

编排：web + worker + beat 一起起来

论滚动发布时别让任务丢

类型注解 / 可观测 / 安全

类型注解与 mypy：给动态语言加护栏

论为什么 Django 项目也值得上 mypy

日志：结构化、带 trace、别打敏感

Metrics：用数字看见系统健康

分布式追踪：一次请求穿过多少服务

论Metrics / 日志 / 追踪 各管一摊

安全底线：CSRF / XSS / SQL 注入 / 依赖

依赖安全：你的第三方库就是攻击面

论安全是持续动作不是一次性

上线开关：DEBUG 与 ALLOWED_HOSTS

下一步往哪走

路学完 Python Web 进阶之后

论Metrics / 日志 / 追踪各管一摊