Skip to content

第四章 架构设计 - 缓存架构

4.1 缓存架构概述

项目使用 Redis 作为缓存后端,通过 Django 的缓存框架实现统一的缓存管理。缓存主要用于以下场景:

+------------------+
|   Django App     |
+------------------+
        |
        v
+------------------+
| Django Cache     |  <-- Django缓存框架
| Framework        |
+------------------+
        |
        v
+------------------+
| django-redis     |  <-- Redis缓存后端
+------------------+
        |
        v
+------------------+
|    Redis         |  <-- Redis服务器
+------------------+

4.2 Redis 配置

缓存后端配置

application/settings.py 中配置 Redis 缓存:

python
CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": f"redis://{env.REDIS_HOST}:{env.REDIS_PORT}/{env.REDIS_DB}",
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
            "PASSWORD": env.REDIS_PASSWORD if env.REDIS_PASSWORD else None,
            "SOCKET_CONNECT_TIMEOUT": 5,  # 连接超时(秒)
            "SOCKET_TIMEOUT": 5,          # 读写超时(秒)
            "CONNECTION_POOL_KWARGS": {"max_connections": 100},  # 连接池大小
            "IGNORE_EXCEPTIONS": True,    # Redis不可用时不抛异常,降级处理
        },
        "KEY_PREFIX": "djangoadmin",  # Key前缀,避免多项目共用Redis时冲突
    }
}

Redis 连接参数

config/env.py 中定义 Redis 连接参数:

python
REDIS_HOST = os.getenv('REDIS_HOST', '127.0.0.1')
REDIS_PORT = int(os.getenv('REDIS_PORT', 6379))
REDIS_PASSWORD = os.getenv('REDIS_PASSWORD', '123456')
REDIS_DB = int(os.getenv('REDIS_DB', 0))
参数默认值说明
REDIS_HOST127.0.0.1Redis服务器地址
REDIS_PORT6379Redis端口
REDIS_PASSWORD123456Redis密码(.env 中配置,无密码时留空)
REDIS_DB0Redis数据库索引(0-15)

4.3 缓存应用场景

4.3.1 Token 黑名单

退出登录时将 Token 加入黑名单,使 Token 立即失效。

实现代码utils/jwt.py):

python
_TOKEN_BLACKLIST_PREFIX = 'token:blacklist:'

def _token_fingerprint(token: str) -> str:
    """生成Token指纹(SHA256摘要)"""
    return hashlib.sha256(token.encode('utf-8')).hexdigest()

def add_token_to_blacklist(token: str) -> bool:
    """将Token加入黑名单"""
    from django_redis import get_redis_connection
    redis = get_redis_connection("default")

    # 获取Token剩余有效期
    expiration = get_token_expiration(token)
    if expiration is None:
        ttl_seconds = DEFAULT_TIMEOUT_MINUTES * 60
    else:
        ttl_seconds = int(
            (expiration - datetime.datetime.now(tz=datetime.timezone.utc)).total_seconds()
        )
        if ttl_seconds <= 0:
            return True  # Token已自然过期,无需加入黑名单

    # 使用SHA256指纹作为Key,避免存储完整Token
    fingerprint = _token_fingerprint(token)
    key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
    redis.set(key, '1', ex=ttl_seconds)

    return True

def is_token_blacklisted(token: str) -> bool:
    """检查Token是否在黑名单中"""
    from django_redis import get_redis_connection
    redis = get_redis_connection("default")

    fingerprint = _token_fingerprint(token)
    key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
    return redis.exists(key) > 0

Key格式djangoadmin:token:blacklist:<sha256_fingerprint>

TTL策略:与 Token 剩余有效期一致,Token 自然过期后黑名单条目自动清除。

降级策略:Redis 不可用时采用 fail-open 策略放行请求,避免全部请求被拒。

4.3.2 IP 地理位置缓存

缓存 IP 地理位置查询结果,避免重复查询外部 API 或本地数据库。

配置项config/env.py):

python
# IP地理位置信息来源:'local' - 本地数据库, 'api' - 在线API
IP_LOCATION_SOURCE = os.environ.get('IP_LOCATION_SOURCE', 'api')

# 是否启用IP地理位置缓存
IP_CACHE_ENABLED = os.environ.get('IP_CACHE_ENABLED', 'True') == 'True'

# IP缓存时间(秒),默认1天
IP_CACHE_TIMEOUT = int(os.environ.get('IP_CACHE_TIMEOUT', 86400))
配置项默认值说明
IP_LOCATION_SOURCEapiIP查询来源:本地数据库或在线API
IP_CACHE_ENABLEDTrue是否启用缓存
IP_CACHE_TIMEOUT86400缓存过期时间(秒),默认1天

4.3.3 验证码存储

图形验证码通过 Redis 存储,支持自动过期。

配置项application/settings.py):

python
CAPTCHA_IMAGE_SIZE = (200, 60)      # 验证码图片尺寸
CAPTCHA_LENGTH = 6                  # 验证码字符个数
CAPTCHA_TIMEOUT = 1                 # 验证码超时时间(分钟)
CAPTCHA_FONT_SIZE = 40              # 验证码字体大小
CAPTCHA_FOREGROUND_COLOR = "#1d953f"  # 验证码前景色
CAPTCHA_BACKGROUND_COLOR = "#ffffff"  # 验证码背景色

4.4 缓存 Key 命名规范

所有缓存 Key 统一使用 djangoadmin: 前缀(通过 KEY_PREFIX 配置),避免多项目共用 Redis 时冲突。

场景Key格式TTL说明
Token黑名单djangoadmin:token:blacklist:<fingerprint>Token剩余有效期SHA256指纹
IP位置缓存djangoadmin:ip:location:<ip>86400秒IP查询结果
验证码djangoadmin:captcha:<key>60秒图形验证码

4.5 缓存降级策略

Redis 不可用时的处理

系统配置了 IGNORE_EXCEPTIONS: True,当 Redis 不可用时:

  1. Token黑名单检查:采用 fail-open 策略,放行请求(已注销的Token可能被误放行)
  2. IP位置缓存:跳过缓存,直接查询数据源
  3. 验证码:降级为内存存储或其他方案
python
# Token黑名单检查的降级处理
def is_token_blacklisted(token: str) -> bool:
    try:
        from django_redis import get_redis_connection
        redis = get_redis_connection("default")
        fingerprint = _token_fingerprint(token)
        key = f'{_TOKEN_BLACKLIST_PREFIX}{fingerprint}'
        return redis.exists(key) > 0
    except Exception as e:
        # fail-open: Redis不可用时放行
        _log_blacklist_check_alert(e)
        return False

告警节流

Redis 故障期间,采用节流告警避免日志刷屏:

python
_BLACKLIST_ALERT_INTERVAL = 60.0  # 告警间隔(秒)
_blacklist_check_last_alert = 0.0

def _log_blacklist_check_alert(error: Exception) -> None:
    """节流记录黑名单检查失败告警"""
    global _blacklist_check_last_alert
    now = time.monotonic()
    if now - _blacklist_check_last_alert >= _BLACKLIST_ALERT_INTERVAL:
        _blacklist_check_last_alert = now
        logger.error("Token黑名单检查异常,已按 fail-open 放行: %s", error)
    else:
        logger.debug("Token黑名单检查异常(节流中,跳过 ERROR): %s", error)

4.6 缓存最佳实践

Key 设计原则

  1. 使用前缀:所有 Key 使用统一前缀,便于管理和清理
  2. 避免冲突:不同业务使用不同的 Key 格式
  3. 合理TTL:根据业务场景设置合适的过期时间
  4. 使用指纹:敏感数据(如Token)使用摘要值作为Key

性能优化

  1. 连接池:配置 max_connections: 100,复用连接
  2. 超时控制:连接和读写超时各5秒,避免长时间阻塞
  3. 批量操作:尽量使用 pipeline 减少网络往返
  4. 降级处理:Redis 不可用时不影响核心业务

安全考虑

  1. 密码保护:Redis 连接使用密码认证
  2. 网络隔离:生产环境 Redis 应限制内网访问
  3. Key加密:敏感数据的 Key 使用 SHA256 摘要
  4. 定期清理:依赖 TTL 自动过期,避免内存溢出

小蚂蚁云团队 · 提供技术支持