在Django应用的高并发场景下，大量重复的数据库查询会快速耗尽数据库连接资源，导致接口响应变慢甚至服务崩溃。使用Redis集群作为二级缓存，可以把高频查询的结果暂存起来，后续相同查询直接从缓存获取，大幅降低数据库访问压力。
Django并发查询的核心痛点
Django默认的数据库查询没有内置的分布式缓存机制，当并发量上升时会出现几个典型问题：
- 相同查询重复执行，浪费数据库计算资源
- 数据库连接池被占满，新请求需要等待连接释放
- 数据库CPU和IO负载过高，整体响应延迟上升
Redis集群作为二级缓存的优势
选择Redis集群而不是单节点Redis，主要是为了解决几个问题：
- 单节点Redis存在容量上限，集群可以横向扩展存储容量
- 单节点故障会导致缓存全部失效，集群有高可用机制
- 集群可以分散请求压力，避免单节点成为性能瓶颈
环境准备与依赖安装
首先需要安装相关的Python依赖包，执行以下命令：
pip install django redis-py-cluster
同时需要提前部署好Redis集群，假设集群节点地址如下：
- 192.168.0.1:6379
- 192.168.0.2:6379
- 192.168.0.3:6379
Redis集群配置
在Django的settings.py文件中添加Redis集群的配置：
# Redis集群配置
REDIS_CLUSTER_NODES = [
("192.168.0.1", 6379),
("192.168.0.2", 6379),
("192.168.0.3", 6379),
]
# 缓存配置
CACHES = {
"default": {
"BACKEND": "django_redis.cache.RedisCache",
"LOCATION": REDIS_CLUSTER_NODES,
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
"CONNECTION_POOL_KWARGS": {"max_connections": 100},
"IGNORE_EXCEPTIONS": True, # 缓存异常时不影响正常业务逻辑
}
}
}
封装二级缓存工具类
为了统一缓存操作，我们可以封装一个工具类，支持缓存的读取、写入和删除：
from django.core.cache import cache
import hashlib
import json
class RedisCacheUtil:
@staticmethod
def get_cache_key(prefix, params):
# 生成唯一的缓存键，避免键冲突
param_str = json.dumps(params, sort_keys=True)
param_hash = hashlib.md5(param_str.encode("utf-8")).hexdigest()
return f"{prefix}:{param_hash}"
@staticmethod
def get(cache_key):
# 从缓存获取数据
return cache.get(cache_key)
@staticmethod
def set(cache_key, value, timeout=300):
# 写入缓存，默认过期时间5分钟
cache.set(cache_key, value, timeout)
@staticmethod
def delete(cache_key):
# 删除缓存
cache.delete(cache_key)
在视图中集成二级缓存
以查询用户列表的接口为例，实现带二级缓存的查询逻辑：
from django.http import JsonResponse
from .models import User
from .utils.redis_cache import RedisCacheUtil
import json
def user_list_view(request):
# 获取查询参数
page = request.GET.get("page", 1)
page_size = request.GET.get("page_size", 10)
# 生成缓存键
cache_key = RedisCacheUtil.get_cache_key(
prefix="user_list",
params={"page": page, "page_size": page_size}
)
# 先查缓存
cache_data = RedisCacheUtil.get(cache_key)
if cache_data:
return JsonResponse({
"code": 0,
"data": cache_data,
"msg": "success from cache"
})
# 缓存不存在，查询数据库
try:
page = int(page)
page_size = int(page_size)
except ValueError:
return JsonResponse({
"code": 400,
"msg": "参数错误"
})
# 数据库查询
start = (page - 1) * page_size
end = start + page_size
users = User.objects.all()[start:end]
user_list = []
for user in users:
user_list.append({
"id": user.id,
"username": user.username,
"email": user.email
})
# 写入缓存
RedisCacheUtil.set(cache_key, user_list, timeout=300)
return JsonResponse({
"code": 0,
"data": user_list,
"msg": "success from db"
})
缓存更新策略
当数据库数据发生变更时，需要及时更新缓存，避免缓存脏数据：
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from .models import User
from .utils.redis_cache import RedisCacheUtil
# 用户数据保存后，删除相关缓存
@receiver(post_save, sender=User)
def user_save_handler(sender, instance, **kwargs):
# 删除用户列表相关的缓存，这里可以根据实际业务删除对应前缀的缓存
cache_key = RedisCacheUtil.get_cache_key(
prefix="user_list",
params={"page": 1, "page_size": 10}
)
RedisCacheUtil.delete(cache_key)
# 用户数据删除后，删除相关缓存
@receiver(post_delete, sender=User)
def user_delete_handler(sender, instance, **kwargs):
cache_key = RedisCacheUtil.get_cache_key(
prefix="user_list",
params={"page": 1, "page_size": 10}
)
RedisCacheUtil.delete(cache_key)
记得在Django的apps.py中注册信号，避免信号不生效：
from django.apps import AppConfig
class UserConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "user"
def ready(self):
import user.signals # 导入信号模块
性能对比测试
我们可以通过简单的压测对比加缓存前后的性能差异，假设使用100并发查询用户列表接口：
| 场景 | 平均响应时间 | 数据库QPS | 接口成功率 |
|---|---|---|---|
| 未加缓存 | 320ms | 1200 | 98% |
| 加Redis集群缓存 | 45ms | 120 | 100% |
从测试结果可以看出，引入Redis集群二级缓存后，接口响应速度提升明显，数据库压力下降了一个数量级。
注意事项
- 缓存过期时间需要根据业务场景设置，高频变更的数据过期时间要短
- 缓存键要做好命名规范，避免不同业务键冲突
- Redis集群节点变更时，要及时更新Django配置中的节点列表
- 缓存操作要加异常处理，避免缓存故障影响正常业务