大数据平台中Django接口的高可用工程实践

1. 项目概述：为什么一个“简单接口”在大数据开发流程里反而最常卡住进度？

你有没有遇到过这样的场景：数据管道跑得飞起，Spark作业调度稳定如钟表，Flink实时流处理延迟压到200毫秒以内，结果前端同学发来一条消息：“后端那个查用户行为统计的接口，点了三次都没反应，能看看吗？”——点开日志，发现不是Kafka消费失败，也不是HDFS权限报错，而是Django视图函数里一个filter()写错了字段名，或者values_list()少了个flat=True，导致ORM生成了全表扫描的SQL，把PostgreSQL拖到CPU 98%。这不是段子，是我在某数据中台项目里连续踩过三次的坑。

“大数据开发之Django简单接口开发”这个标题看似平平无奇，但恰恰戳中了当前数据工程落地中最隐蔽的断点：数据能力最终必须以API为出口，而API的健壮性、可观测性、可维护性，从来不是“附带任务”，而是数据价值能否被业务真正用起来的生死线。它不涉及Hadoop调优或Flink状态后端选型，却直接决定数据产品上线周期是3天还是3周；它不需要你手写UDF，但要求你对Django ORM的查询计划、数据库连接池行为、异步任务与同步响应的边界有肌肉记忆般的直觉。

我把它称为“最后一公里工程”——上游所有复杂的数据清洗、特征计算、模型训练，最终都得靠这几个HTTP端点“交卷”。而所谓“简单”，往往是最危险的幻觉。一个返回5个字段的GET接口，背后可能串联着ClickHouse聚合查询、Redis缓存穿透防护、Elasticsearch多条件检索、以及对下游微服务的gRPC调用。它之所以被标记为“简单”，只是因为没写在技术方案PPT的架构图里，而不是因为它真的轻量。

这篇文章就是为你拆解：当你的核心工作是构建数据底座时，如何让这些“配角级”的Django接口，既扛得住每秒300+ QPS的报表导出请求，又能在凌晨三点精准报警出问题模块，还能让新来的实习生看一眼代码就知道缓存键怎么生成、错误码怎么定义、性能瓶颈在哪一行。我会从零开始，不假设你熟悉Django高级特性，但会默认你了解SQL基本语法和Linux基础命令——毕竟，你是个数据工程师，不是刚毕业的Python新手。

2. 整体设计思路：为什么不用FastAPI？为什么坚持用Django？三个被低估的硬性约束

很多人看到标题第一反应是：“都2024年了，还用Django写API？FastAPI不是更快更现代？”这个问题我问过自己不下十次。在某次A/B测试中，我们真用FastAPI重写了同一组接口，压测结果显示QPS高了17%，序列化快了22%，但上线后第三天就回滚了。原因不在性能，而在三个被严重低估的硬性约束——它们不是技术选型文档里的虚词，而是深夜救火时的真实痛感。

2.1 约束一：团队技能栈的“最小公分母”必须是Django

我们团队有6个数据工程师，其中4个主攻Spark SQL和Airflow DAG编写，2个偏重实时计算。但所有人——包括刚转岗过来的DBA——都至少维护过一个Django后台管理系统。这意味着：

• 当某个报表接口需要紧急加一个“按地域维度下钻”的参数时，值班的数据同学能直接改views.py，不用等后端同学排期；
• 当线上出现慢查询，大家能立刻看懂django.db.connection.queries打印的SQL，而不是对着FastAPI的@router.get装饰器发懵；
• 更关键的是，Django Admin自带的数据库记录增删改查界面，让数据QA同学能自己验证接口返回值是否与底层表一致，把“接口逻辑正确性”这个黑盒，变成了可触摸的白盒。

FastAPI的Pydantic模型确实优雅，但当你需要快速给一个临时分析需求加接口时，Django的ModelSerializer配合ModelViewSet，三行代码就能跑通CRUD，而FastAPI要写schema、写router、写dependency、写exception handler——这多出来的15分钟，在数据平台迭代节奏里，就是当天能不能交付的分水岭。

2.2 约束二：数据权限体系必须与Django Auth深度耦合

大数据场景下的接口，90%以上需要做细粒度权限控制。比如：

• 某销售部门只能查本区域的客户转化漏斗；
• 数据分析师能看到全量脱敏字段，但不能导出原始手机号；
• 运营同学能调用实时UV接口，但不能触发离线特征重算任务。

Django的User、Group、Permission模型，配合django-guardian的ObjectPermission，已经形成一套成熟、可审计、可配置的权限框架。我们直接复用这套体系，把权限规则写进get_queryset()方法里：

def get_queryset(self): # 自动过滤：普通用户只能看到自己创建的报表 qs = super().get_queryset() if not self.request.user.is_superuser: qs = qs.filter(created_by=self.request.user) # 再叠加数据域权限：销售总监能看到全国，区域经理只能看本省 data_scope = self.request.user.profile.data_scope if data_scope == 'province': qs = qs.filter(province=self.request.user.profile.province_code) return qs

换成FastAPI？你得自己实现JWT解析、RBAC策略引擎、对象级权限中间件——而这些轮子，Django已经帮你造好十年了，且经过百万级生产环境验证。在数据平台这种“稳定压倒一切”的场景里，重复造轮子不是技术自信，是给自己埋雷。

2.3 约束三：监控告警必须无缝接入现有运维体系

我们所有的服务指标（CPU、内存、HTTP状态码）都通过Prometheus+Grafana统一采集。Django生态有成熟的django-prometheus包，只需两行配置，就能自动暴露django_http_requests_total、django_http_request_duration_seconds等标准指标。而FastAPI的Prometheus集成，需要手动注册middleware、处理异步上下文、适配ASGI生命周期——在我们已有的Ansible部署脚本里，Django的监控接入是“开箱即用”，FastAPI则要额外写200行胶水代码。

这三个约束，不是技术优劣的辩论题，而是数据工程师每天面对的现实：你写的不是炫技Demo，而是要和Airflow、Trino、Superset共存三年以上的生产系统。Django的“重”在这里反而是优势——它的约定大于配置、它的生态足够厚重、它的学习曲线虽然陡峭但一旦掌握就极少踩坑。所以，我们坚持用Django，不是守旧，而是选择了一条“故障率最低、交接成本最小、长期维护最省心”的路径。

3. 核心细节解析：从“能跑”到“稳跑”的五个关键实操节点

很多数据工程师第一次写Django接口，目标很朴素：让curl http://localhost:8000/api/v1/users/返回JSON就行。但真实生产环境里，“能跑”和“稳跑”之间，隔着五个必须亲手填平的坑。下面是我整理的、每个节点都附带真实日志截图和修复对比的实操清单。

3.1 节点一：数据库连接池——别让一个接口拖垮整个PostgreSQL

现象：接口平均响应时间从200ms突然飙升到3秒，pg_stat_activity显示大量idle in transaction状态的连接。

根因：Django默认使用CONN_MAX_AGE=0，每次HTTP请求都新建数据库连接，用完立即关闭。但在高并发下，频繁建连/断连消耗巨大，且PostgreSQL连接数有限（默认100），很容易被占满。

解决方案：启用持久化连接池，并精确控制大小。

# settings.py DATABASES = { 'default': { 'ENGINE': 'django.db.backends.postgresql', 'NAME': 'data_platform', 'USER': 'dp_app', 'PASSWORD': os.getenv('DB_PASSWORD'), 'HOST': 'pg-prod.internal', 'PORT': '5432', 'CONN_MAX_AGE': 60, # 连接复用60秒 'OPTIONS': { 'MAX_CONNS': 20, # 这个参数需配合django-db-geventpool使用 } } }

但注意：CONN_MAX_AGE只是第一步。真正的连接池管理，需要引入django-db-geventpool（适配Gunicorn+gevent部署模式）或dj-database-url（适配连接字符串）。我们实测：未启用连接池时，100并发下PostgreSQL连接数峰值达92；启用后稳定在23个，且响应时间标准差从±1200ms降到±80ms。

提示：别盲目调大MAX_CONNS。我们曾设为50，结果发现PostgreSQL服务器内存暴涨，因为每个连接固定占用约10MB内存。计算公式：MAX_CONNS × 10MB < 服务器可用内存 × 0.7。你得先算清楚账。

3.2 节点二：ORM查询优化——警惕select_related和prefetch_related的误用

现象：一个返回100条订单记录的接口，执行时间长达8秒，EXPLAIN ANALYZE显示嵌套循环JOIN了5张表。

根因：Django ORM的懒加载（lazy loading）机制。当你在模板里写{{ order.customer.name }}时，如果没提前关联，Django会在循环里为每条订单发起一次SELECT * FROM customer WHERE id = ?查询——N+1问题。

错误写法：

# views.py —— 千万别这么写！ orders = Order.objects.filter(status='shipped') # 模板里遍历时会触发100次customer查询

正确写法（根据关联类型选择）：

# 一对一或外键关联（正向）→ 用 select_related orders = Order.objects.select_related('customer', 'address').filter(status='shipped') # 多对多或反向外键（一对多）→ 用 prefetch_related orders = Order.objects.prefetch_related('items__product').filter(status='shipped') # 复杂场景：混合使用 + 显式指定字段 orders = Order.objects.select_related('customer').prefetch_related( models.Prefetch( 'items', queryset=OrderItem.objects.select_related('product').only('id', 'quantity', 'product__name') ) ).filter(status='shipped').only('id', 'order_no', 'customer__email')

关键技巧：only()和defer()不是可选项，是必选项。我们有个接口原本返回32个字段，加上关联表后SQL扫描行数超200万；加上only('id', 'name', 'status')后，扫描行数降到1.2万，耗时从8秒压到320毫秒。

3.3 节点三：缓存策略——Redis不是万能药，得知道缓存什么、缓存多久

现象：用户反馈“刚更新的报表数据，刷新页面还是旧的”，而运维告警显示Redis内存使用率95%。

根因：缓存滥用。把所有接口都套上@cache_page(60)，结果缓存了大量个性化数据（如用户专属看板），既浪费内存，又导致数据不一致。

我们的缓存分层策略：

实操代码示例（L2查询级缓存）：

from django.core.cache import cache import hashlib def stats_api(request): # 用请求参数生成唯一缓存键 cache_key = f"stats_{hashlib.md5(request.GET.urlencode().encode()).hexdigest()}" cached_data = cache.get(cache_key) if cached_data is not None: return JsonResponse(cached_data) # 执行实际查询（这里可能是调用Trino或ClickHouse） data = execute_analytics_query(request.GET.dict()) # 设置缓存，注意：只缓存计算结果，不缓存整个HttpResponse cache.set(cache_key, data, timeout=180) return JsonResponse(data)

注意：缓存键必须包含所有影响结果的参数。我们曾漏掉timezone参数，导致不同时区用户看到同一份缓存数据，引发客诉。

3.4 节点四：错误处理与日志——别让500错误变成“接口挂了”的模糊描述

现象：前端报“网络错误”，但Nginx日志显示200，Django日志却空空如也。

根因：未统一异常捕获。Django默认把ValueError、KeyError等抛到WSGI层，被Gunicorn吞掉，只留下Internal Server Error。

我们的标准化错误响应结构：

{ "code": 40001, "message": "日期格式错误，请使用YYYY-MM-DD格式", "request_id": "req_abc123xyz", "timestamp": "2024-05-20T14:22:33Z" }

实现方式（自定义异常中间件）：

# middleware.py class APIExceptionMiddleware: def __init__(self, get_response): self.get_response = get_response def __call__(self, request): return self.get_response(request) def process_exception(self, request, exception): # 生成唯一request_id，贯穿整个请求链路 request_id = str(uuid.uuid4()) logger.error(f"API Exception: {exception}", extra={'request_id': request_id}) if isinstance(exception, ValidationError): return JsonResponse({ 'code': 40001, 'message': str(exception), 'request_id': request_id, 'timestamp': timezone.now().isoformat() }, status=400) elif isinstance(exception, DatabaseError): logger.critical("Database error occurred", extra={'request_id': request_id}) return JsonResponse({ 'code': 50001, 'message': '数据服务暂时不可用，请稍后重试', 'request_id': request_id, 'timestamp': timezone.now().isoformat() }, status=500) # 其他异常...

关键心得：日志级别必须严格区分。logger.error()用于用户可见错误（如参数校验失败），logger.critical()用于必须人工介入的故障（如数据库连接池耗尽）。我们用ELK收集日志，设置告警规则：critical日志每分钟超过3条，立即电话通知。

3.5 节点五：部署与健康检查——K8s里的livenessProbe不是摆设

现象：K8s集群频繁重启Pod，但kubectl logs显示一切正常。

根因：健康检查探针配置不当。我们最初用httpGet探针检测/healthz，但该接口只检查Django能否启动，不检查数据库连接、Redis连通性、外部服务可用性。

改进后的/healthz实现：

# views.py def health_check(request): checks = { 'django': True, 'database': False, 'redis': False, 'clickhouse': False, } # 检查数据库（执行轻量查询） try: from django.db import connection with connection.cursor() as cursor: cursor.execute("SELECT 1") checks['database'] = cursor.fetchone()[0] == 1 except Exception as e: logger.warning(f"Database health check failed: {e}") # 检查Redis try: from django.core.cache import cache cache.set('health_test', 'ok', timeout=1) checks['redis'] = cache.get('health_test') == 'ok' except Exception as e: logger.warning(f"Redis health check failed: {e}") # 检查ClickHouse（调用其system.ping） try: import requests resp = requests.get("http://clickhouse-prod:8123/ping", timeout=2) checks['clickhouse'] = resp.status_code == 200 except Exception as e: logger.warning(f"ClickHouse health check failed: {e}") status_code = 200 if all(checks.values()) else 503 return JsonResponse({'status': 'ok' if status_code == 200 else 'unhealthy', 'checks': checks}, status=status_code)

K8s配置同步更新：

livenessProbe: httpGet: path: /healthz port: 8000 initialDelaySeconds: 30 periodSeconds: 10 timeoutSeconds: 5 failureThreshold: 3 # 连续3次失败才重启

效果：Pod重启频率从每天12次降到每月1次。因为现在探针能真实反映服务“是否可用”，而不是“是否活着”。

4. 实操过程详解：从零搭建一个支持千万级用户行为查询的Django接口

现在，我们动手实现一个真实场景：提供一个API，允许运营同学按日期、设备类型、渠道来源，查询指定时间段内的用户行为事件总数（如点击、曝光、购买）。这个接口要支撑每日5000万事件的查询压力，且响应时间<800ms。

4.1 步骤一：数据源选型与模型设计——为什么不用Django Model直接映射事件表？

首先明确：千万级行为事件表，绝不能用Django Model做ORM操作。Event.objects.filter(date='2024-05-20', device='ios').count()这种写法，在PostgreSQL上会触发全表扫描，哪怕加了索引，单次查询也要20秒。

我们的分层架构：

• 热数据（最近7天）：存于ClickHouse，利用其列式存储和向量化执行引擎；
• 温数据（7-90天）：存于PostgreSQL分区表，按date字段月度分区；
• 冷数据（90天以上）：归档至对象存储，仅支持离线分析。

因此，Django Model只定义元数据（如渠道配置、设备字典），不定义事件本身：

# models.py class Channel(models.Model): """渠道配置表，供前端下拉选择""" code = models.CharField(max_length=32, unique=True) # wechat, ios_app, android_app name = models.CharField(max_length=64) is_active = models.BooleanField(default=True) class DeviceType(models.Model): """设备类型字典""" code = models.CharField(max_length=16, unique=True) # ios, android, web name = models.CharField(max_length=32) # 事件表不在此定义！由ClickHouse和PostgreSQL原生SQL查询

4.2 步骤二：查询引擎封装——抽象出统一的查询接口

创建analytics/query_engine.py，屏蔽底层数据源差异：

from typing import Dict, List, Any import logging logger = logging.getLogger(__name__) class QueryEngine: @staticmethod def query_events(filters: Dict[str, Any]) -> Dict[str, int]: """ 统一查询入口 filters示例: {'date_from': '2024-05-01', 'date_to': '2024-05-20', 'device': 'ios', 'channel': 'wechat'} 返回: {'click': 123456, 'expose': 789012, 'purchase': 54321} """ # 1. 判断时间范围，选择数据源 date_from = filters.get('date_from') date_to = filters.get('date_to') if not date_from or not date_to: raise ValueError("date_from and date_to are required") # 计算时间跨度（天数） from datetime import datetime days_span = (datetime.strptime(date_to, '%Y-%m-%d') - datetime.strptime(date_from, '%Y-%m-%d')).days + 1 if days_span <= 7: # 热数据 → ClickHouse return QueryEngine._query_clickhouse(filters) elif days_span <= 90: # 温数据 → PostgreSQL return QueryEngine._query_postgresql(filters) else: # 冷数据 → 返回错误提示，引导走离线任务 raise ValueError("查询时间跨度超过90天，请使用离线分析功能") @staticmethod def _query_clickhouse(filters: Dict[str, Any]) -> Dict[str, int]: # 使用clickhouse-driver执行原生SQL from clickhouse_driver import Client client = Client(host='clickhouse-prod.internal') sql = """ SELECT event_type, count(*) as cnt FROM user_events WHERE toDate(event_time) BETWEEN %(date_from)s AND %(date_to)s """ params = {'date_from': filters['date_from'], 'date_to': filters['date_to']} if filters.get('device'): sql += " AND device = %(device)s" params['device'] = filters['device'] if filters.get('channel'): sql += " AND channel = %(channel)s" params['channel'] = filters['channel'] sql += " GROUP BY event_type" try: results = client.execute(sql, params) return {row[0]: row[1] for row in results} except Exception as e: logger.error(f"ClickHouse query failed: {e}", exc_info=True) raise @staticmethod def _query_postgresql(filters: Dict[str, Any]) -> Dict[str, int]: # 使用Django ORM的raw()方法，避免ORM开销 from django.db import connection with connection.cursor() as cursor: sql = """ SELECT event_type, count(*) as cnt FROM user_events_p2024_05 -- 动态选择分区表名 WHERE event_date BETWEEN %(date_from)s AND %(date_to)s """ params = {'date_from': filters['date_from'], 'date_to': filters['date_to']} # ... 同上添加device/channel条件 cursor.execute(sql, params) return {row[0]: row[1] for row in cursor.fetchall()}

4.3 步骤三：视图实现与性能压测——如何把800ms的P95延迟压到320ms

# views.py from django.http import JsonResponse from django.views.decorators.csrf import csrf_exempt from django.utils.decorators import method_decorator from django.views import View from analytics.query_engine import QueryEngine from django.core.cache import cache import hashlib import json @method_decorator(csrf_exempt, name='dispatch') class EventStatsView(View): def post(self, request): try: # 1. 解析JSON请求体 data = json.loads(request.body) filters = data.get('filters', {}) # 2. 参数校验（比Django REST Framework更轻量） required_fields = ['date_from', 'date_to'] for field in required_fields: if not filters.get(field): return JsonResponse({ 'code': 40001, 'message': f'{field} is required' }, status=400) # 3. 生成缓存键（包含所有影响结果的参数） cache_key = f"event_stats_{hashlib.md5(json.dumps(filters, sort_keys=True).encode()).hexdigest()}" # 4. 尝试读缓存 cached_result = cache.get(cache_key) if cached_result: return JsonResponse({ 'code': 0, 'data': cached_result, 'cached': True, 'timestamp': timezone.now().isoformat() }) # 5. 执行查询（核心耗时步骤） result = QueryEngine.query_events(filters) # 6. 写缓存（注意：只缓存结果，不缓存整个Response） cache.set(cache_key, result, timeout=180) # 3分钟 return JsonResponse({ 'code': 0, 'data': result, 'cached': False, 'timestamp': timezone.now().isoformat() }) except json.JSONDecodeError: return JsonResponse({'code': 40002, 'message': 'Invalid JSON'}, status=400) except ValueError as e: return JsonResponse({'code': 40003, 'message': str(e)}, status=400) except Exception as e: logger.exception("Unexpected error in EventStatsView") return JsonResponse({'code': 50000, 'message': 'Internal server error'}, status=500)

压测结果对比（Locust工具，100并发持续5分钟）：

关键技巧：不要迷信“缓存能解决一切”。我们发现，当缓存命中率低于60%时（比如运营同学总在查不同日期组合），缓存反而增加Redis负载。所以我们在监控面板里加了“缓存命中率”指标，低于70%就触发告警，提醒优化查询参数设计。

4.4 步骤四：安全加固——防刷、防爬、防越权的三道防线

大数据接口是攻击者眼中的“金矿”，因为返回的数据往往高价值、低防护。我们部署了三层防御：

第一道：速率限制（Rate Limiting）
使用django-ratelimit，按IP+用户ID双重限制：

from django_ratelimit.decorators import ratelimit @ratelimit(key='ip', rate='100/m', method='POST', block=True) @ratelimit(key='user_or_ip', rate='500/h', method='POST', block=True) def post(self, request): # 视图逻辑

• 普通用户：每分钟最多100次请求（防脚本暴力探测）
• 登录用户：每小时最多500次（防账号被盗用）
• 超限后返回429 Too Many Requests，并记录到审计日志

第二道：参数签名（Signature）
对敏感查询（如导出全量数据），要求客户端用私钥对参数签名：

# 客户端伪代码 params = {'date_from': '2024-05-01', 'date_to': '2024-05-20'} signature = hmac.new(SECRET_KEY, json.dumps(params, sort_keys=True).encode(), 'sha256').hexdigest() # 发送 headers: X-Signature: signature

服务端验证：

def verify_signature(request, params): signature = request.headers.get('X-Signature') expected = hmac.new(SECRET_KEY, json.dumps(params, sort_keys=True).encode(), 'sha256').hexdigest() return hmac.compare_digest(signature, expected)

第三道：数据脱敏（Data Masking）
在序列化阶段动态脱敏：

def serialize_event_data(data): if not request.user.has_perm('analytics.view_raw_data'): # 对手机号、身份证号等字段进行掩码 for item in data.get('events', []): if 'phone' in item: item['phone'] = item['phone'][:3] + '****' + item['phone'][-4:] if 'id_card' in item: item['id_card'] = item['id_card'][:4] + '********' + item['id_card'][-4:] return data

5. 常见问题与排查技巧实录：那些让你凌晨三点爬起来的“幽灵Bug”

最后，分享几个我在生产环境里反复遭遇、且文档几乎不提的“幽灵Bug”。它们不常发生，但一旦出现，排查时间远超修复时间。

5.1 问题一：Django Admin里能查到数据，API却返回空列表——时间时区的陷阱

现象：运营同学在Django Admin里看到2024-05-20的订单，但调用/api/v1/orders/?date=2024-05-20返回空数组。

根因：Django的DateTimeField在数据库里存的是UTC时间，而Admin界面默认按TIME_ZONE = 'Asia/Shanghai'渲染，API却按UTC解析date参数。

排查步骤：

1. 查看Django设置：settings.TIME_ZONE和settings.USE_TZ
2. 检查数据库字段类型：event_time是TIMESTAMP WITH TIME ZONE还是TIMESTAMP WITHOUT TIME ZONE？
3. 在Django shell里执行：

   from django.utils import timezone print(timezone.now()) # 输出UTC时间 print(timezone.localtime()) # 输出本地时间

解决方案：统一用timezone.localtime()处理前端传入的日期：

from django.utils import timezone def get_queryset(self): date_str = self.request.GET.get('date') if date_str: # 将字符串解析为本地时间，再转为UTC存入查询 local_dt = timezone.make_aware( datetime.strptime(date_str, '%Y-%m-%d'), timezone.get_current_timezone() ) utc_dt = local_dt.astimezone(timezone.utc) # 查询UTC时间范围 return super().get_queryset().filter( event_time__date=utc_dt.date() ) return super().get_queryset()

5.2 问题二：Gunicorn worker进程内存持续增长，最终OOM被K8s杀死

现象：Pod内存使用率从30%缓慢爬升到95%，然后被OOMKilled，日志里只有Killed process。

根因：Django的DEBUG=True模式下，会缓存所有SQL查询，且不释放。即使你在线上关了DEBUG，某些第三方库（如django-debug-toolbar）的残留配置仍会触发内存泄漏。

排查命令：

# 进入Pod，查看内存占用最多的Python对象 pip install pympler python -c " from pympler import tracker tr = tracker.SummaryTracker() tr.print_diff() "

解决方案：

• 线上环境DEBUG=False，且ALLOWED_HOSTS必须显式配置；
• 删除所有debug_toolbar相关代码和中间件；
• 在Gunicorn配置中强制worker重启：

  # gunicorn.conf.py max_requests = 1000 # 每个worker处理1000个请求后重启 max_requests_jitter = 100 # 避免所有worker同时重启

5.3 问题三：Redis缓存击穿——热点Key失效瞬间，大量请求打穿数据库

现象：某个爆款活动期间，/api/v1/promo/stats接口在整点时刻（缓存批量过期）出现雪崩，PostgreSQL CPU飙到100%。

根因：所有请求在同一秒内发现缓存失效，全部涌向数据库。

解决方案：二级缓存 + 逻辑过期：

def get_cached_stats(key, db_fetch_func, expire=180): # 一级缓存：带逻辑过期时间的Redis cached = cache.get(key) if cached and cached.get('expired_at', 0) > time.time(): return cached['data'] # 二级缓存：尝试获取锁，只有一个请求去查DB lock_key = f"lock:{key}" if cache.add(lock_key, "1", timeout=5): # 设置5秒锁 try: data = db_fetch_func() # 写入带逻辑过期的缓存（物理过期设长，逻辑过期设短） cache.set(key, { 'data': data, 'expired_at': time.time() + 60 # 逻辑过期60秒 }, timeout=300) # 物理过期5分钟 return data finally: cache.delete(lock_key) else: # 获取锁失败，等待100ms后重试（避免死等） time.sleep(0.1) return get_cached_stats(key, db_fetch_func, expire)

5.4 问题四：Docker镜像体积爆炸——一个Django应用打包后竟有1.2GB

现象：CI/CD流水线构建Docker镜像耗时12分钟，推送镜像到仓库要8分钟。

根因：基础镜像选错 + 未清理构建缓存。

错误做法：

FROM python:3.9 COPY requirements.txt . RUN pip install -r requirements.txt # 安装了所有dev依赖 COPY . .

正确做法（多阶段构建）：

# 构建阶段 FROM python:3.9-slim AS builder WORKDIR /app COPY requirements.txt . # 只安装生产依赖 RUN pip install --no-cache-dir --user -r requirements.txt # 运行阶段 FROM python:3.9-slim WORKDIR /app # 复制构建好的包，不复制源码 COPY --from=builder /root/.local /root/.local ENV PATH=/root/.local/bin:$PATH COPY . . CMD ["