Become a sponsor

项目使用 Redis 作为缓存后端,通过 Django 的缓存框架实现统一的缓存管理。缓存主要用于以下场景:
+------------------+
| Django App |
+------------------+
|
v
+------------------+
| Django Cache | <-- Django缓存框架
| Framework |
+------------------+
|
v
+------------------+
| django-redis | <-- Redis缓存后端
+------------------+
|
v
+------------------+
| Redis | <-- Redis服务器
+------------------+在 application/settings.py 中配置 Redis 缓存:
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时冲突
}
}在 config/env.py 中定义 Redis 连接参数:
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_HOST | 127.0.0.1 | Redis服务器地址 |
REDIS_PORT | 6379 | Redis端口 |
REDIS_PASSWORD | 123456 | Redis密码(.env 中配置,无密码时留空) |
REDIS_DB | 0 | Redis数据库索引(0-15) |
退出登录时将 Token 加入黑名单,使 Token 立即失效。
实现代码(utils/jwt.py):
_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) > 0Key格式:djangoadmin:token:blacklist:<sha256_fingerprint>
TTL策略:与 Token 剩余有效期一致,Token 自然过期后黑名单条目自动清除。
降级策略:Redis 不可用时采用 fail-open 策略放行请求,避免全部请求被拒。
缓存 IP 地理位置查询结果,避免重复查询外部 API 或本地数据库。
配置项(config/env.py):
# 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_SOURCE | api | IP查询来源:本地数据库或在线API |
IP_CACHE_ENABLED | True | 是否启用缓存 |
IP_CACHE_TIMEOUT | 86400 | 缓存过期时间(秒),默认1天 |
图形验证码通过 Redis 存储,支持自动过期。
配置项(application/settings.py):
CAPTCHA_IMAGE_SIZE = (200, 60) # 验证码图片尺寸
CAPTCHA_LENGTH = 6 # 验证码字符个数
CAPTCHA_TIMEOUT = 1 # 验证码超时时间(分钟)
CAPTCHA_FONT_SIZE = 40 # 验证码字体大小
CAPTCHA_FOREGROUND_COLOR = "#1d953f" # 验证码前景色
CAPTCHA_BACKGROUND_COLOR = "#ffffff" # 验证码背景色所有缓存 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秒 | 图形验证码 |
系统配置了 IGNORE_EXCEPTIONS: True,当 Redis 不可用时:
# 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 FalseRedis 故障期间,采用节流告警避免日志刷屏:
_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)max_connections: 100,复用连接pipeline 减少网络往返