一、为什么使用Redis缓存?
在Django项目中,Redis缓存可以:
- 提升性能:减少数据库查询
- 降低负载:分担数据库压力
- 共享会话:适合多服务器部署
- 存储临时数据:如验证码、短信限制
二、快速开始
1. 安装Redis服务器
Ubuntu/Debian:
sudoaptupdatesudoaptinstallredis-serversudosystemctl start redissudosystemctlenableredis检查Redis是否运行:
redis-cliping# 返回 PONG 表示成功2. 安装Python包
pipinstalldjango-redis redis3. 基本配置
在settings.py中添加:
# 缓存配置CACHES={'default':{'BACKEND':'django_redis.cache.RedisCache','LOCATION':'redis://127.0.0.1:6379/0',# 0号数据库'OPTIONS':{'CLIENT_CLASS':'django_redis.client.DefaultClient',},'TIMEOUT':60*15,# 默认15分钟过期}}# 可选:将Session也存到RedisSESSION_ENGINE='django.contrib.sessions.backends.cache'SESSION_CACHE_ALIAS='default'三、基本使用
1. 简单缓存操作
from django.core.cacheimportcache from django.httpimportJsonResponseimporttimedef cache_demo(request):"""基本缓存操作示例"""# 1. 设置缓存(缓存5分钟)cache.set('username','张三',timeout=300)# 2. 获取缓存username=cache.get('username')# 如果不存在,返回默认值username=cache.get('username','默认用户')# 3. 设置多个值cache.set_many({'key1':'value1','key2':'value2','key3':'value3'})# 4. 获取多个值values=cache.get_many(['key1','key2','key3'])# 5. 删除缓存cache.delete('key1')cache.delete_many(['key2','key3'])# 6. 只有键不存在时才设置cache.add('counter',1)# 第一次设置成功cache.add('counter',2)# 不会覆盖,counter还是1# 7. 数字递增递减cache.set('visits',0)cache.incr('visits')# visits = 1cache.incr('visits',10)# visits = 11cache.decr('visits',5)# visits = 6# 8. 清空所有缓存(谨慎使用)# cache.clear()returnJsonResponse({'username':username})2. 缓存数据库查询结果
from django.core.cacheimportcache from django.httpimportJsonResponse from .modelsimportArticleimportjson def get_articles(request):""" 获取文章列表,使用缓存减少数据库查询""" cache_key='article_list'# 缓存键articles=cache.get(cache_key)ifarticles is None:# 缓存中没有,从数据库查询print("从数据库查询文章...")articles=list(Article.objects.filter(is_published=True).values('id','title','created_at')[:10])# 将结果存入缓存,有效期5分钟cache.set(cache_key, articles,timeout=300)else: print("从缓存获取文章...")returnJsonResponse({'articles':articles})3. 缓存用户数据
from django.contrib.auth.decoratorsimportlogin_required from django.core.cacheimportcache from django.httpimportJsonResponse @login_required def get_user_profile(request):""" 获取用户资料,带用户ID的缓存""" user_id=request.user.id cache_key=f'user_profile:{user_id}'# 尝试从缓存获取profile=cache.get(cache_key)ifprofile is None:# 模拟获取用户资料(数据库查询、API调用等)print(f"查询用户 {user_id} 的资料...")profile={'id':user_id,'username':request.user.username,'email':request.user.email,'last_login':request.user.last_login.isoformat()ifrequest.user.last_loginelseNone}# 缓存用户资料,1小时cache.set(cache_key, profile,timeout=3600)returnJsonResponse(profile)四、视图缓存
1. 缓存整个视图
from django.views.decorators.cacheimportcache_page from django.utils.decoratorsimportmethod_decorator from django.viewsimportView from django.httpimportJsonResponseimporttime# 函数视图缓存@cache_page(60*5)# 缓存5分钟def expensive_view(request):"""耗时的视图,缓存5分钟""" time.sleep(2)# 模拟耗时操作returnJsonResponse({'data':'耗时操作结果','time':time.time()})# 类视图缓存class CachedView(View): @method_decorator(cache_page(60*10))# 缓存10分钟def dispatch(self, *args, **kwargs):returnsuper().dispatch(*args, **kwargs)def get(self, request): time.sleep(1)# 模拟耗时操作returnJsonResponse({'message':'类视图缓存示例'})2. 不缓存特定视图
from django.views.decorators.cacheimportnever_cache from django.httpimportJsonResponse @never_cache def sensitive_data_view(request):"""敏感数据,不缓存"""returnJsonResponse({'sensitive':'这是敏感数据'})五、模板缓存
在模板中使用缓存:
<!-- article_detail.html -->{% load cache %}<h1>{{article.title}}</h1><!-- 缓存侧边栏,5分钟 -->{% cache300sidebar %}<divclass="sidebar"><h3>热门文章</h3><ul>{%forhot_articleinhot_articles %}<li>{{hot_article.title}}</li>{% endfor %}</ul></div>{% endcache %}<!-- 根据用户缓存,不同用户看到不同内容 -->{% cache300user_specific request.user.id %}<divclass="user-info"><h3>欢迎{{request.user.username}}</h3><!-- 用户特定内容 --></div>{% endcache %}<!-- 根据文章ID和更新时间缓存 -->{% cache300article_content article.id article.updated_at %}<divclass="content">{{article.content|safe}}</div>{% endcache %}六、实战案例:商品列表页
1. 模型
# models.pyfrom django.dbimportmodels class Product(models.Model): name=models.CharField(max_length=100)price=models.DecimalField(max_digits=10,decimal_places=2)stock=models.IntegerField(default=0)created_at=models.DateTimeField(auto_now_add=True)updated_at=models.DateTimeField(auto_now=True)def __str__(self):returnself.name2. 带缓存的商品服务
# services/product_service.pyfrom django.core.cacheimportcache from .modelsimportProductimportjson from datetimeimportdatetime class ProductService:"""商品服务,包含缓存逻辑""" @staticmethod def get_products(category=None,limit=20):""" 获取商品列表,带缓存"""# 生成缓存键cache_key='products'ifcategory: cache_key=f'products:{category}'# 尝试从缓存获取products=cache.get(cache_key)ifproducts is None:# 查询数据库queryset=Product.objects.filter(stock__gt=0)ifcategory: queryset=queryset.filter(category=category)products=list(queryset.values('id','name','price','stock')[:limit])# 存入缓存,10分钟cache.set(cache_key, products,timeout=600)returnproducts @staticmethod def get_product_detail(product_id):""" 获取商品详情,带缓存""" cache_key=f'product_detail:{product_id}'product=cache.get(cache_key)ifproduct is None: try: product_obj=Product.objects.get(id=product_id)product={'id':product_obj.id,'name':product_obj.name,'price':float(product_obj.price),'stock':product_obj.stock,'description':getattr(product_obj,'description',''),}# 缓存1小时cache.set(cache_key, product,timeout=3600)except Product.DoesNotExist: product=Nonereturnproduct @staticmethod def clear_product_cache(product_id=None):""" 清理商品缓存"""ifproduct_id:# 删除单个商品缓存cache.delete(f'product_detail:{product_id}')# 删除商品列表缓存cache.delete('products')# 可以添加更多商品相关的缓存清理3. 视图
# views.pyfrom django.httpimportJsonResponse from django.views.decorators.cacheimportcache_page from .services.product_serviceimportProductService @cache_page(60*5)# 缓存5分钟def product_list(request):"""商品列表页""" category=request.GET.get('category')products=ProductService.get_products(category=category)returnJsonResponse({'products':products})def product_detail(request, product_id):"""商品详情页""" product=ProductService.get_product_detail(product_id)ifproduct is None:returnJsonResponse({'error':'商品不存在'},status=404)returnJsonResponse(product)# 后台管理:更新商品时清理缓存def update_product(request, product_id):""" 更新商品(示例,实际需要POST请求和表单验证)"""# ... 更新商品的逻辑 ...# 更新后清理缓存ProductService.clear_product_cache(product_id)returnJsonResponse({'message':'更新成功'})七、高级技巧
1. 缓存装饰器
# utils/decorators.pyfrom django.core.cacheimportcache from functoolsimportwrapsimporthashlibimportjson def cache_result(timeout=300):""" 缓存函数结果的装饰器""" def decorator(func): @wraps(func)def wrapper(*args, **kwargs):# 生成唯一的缓存键func_name=func.__name__ args_str=json.dumps(args,sort_keys=True)kwargs_str=json.dumps(kwargs,sort_keys=True)key_str=f"{func_name}:{args_str}:{kwargs_str}"# 如果key太长,使用hashiflen(key_str)>200: key_hash=hashlib.md5(key_str.encode()).hexdigest()cache_key=f"cache:{func_name}:{key_hash}"else: cache_key=f"cache:{key_str}"# 尝试从缓存获取result=cache.get(cache_key)ifresult is not None:returnresult# 执行函数result=func(*args, **kwargs)# 存入缓存cache.set(cache_key, result,timeout=timeout)returnresultreturnwrapperreturndecorator# 使用示例from utils.decoratorsimportcache_result class UserService: @staticmethod @cache_result(timeout=600)# 缓存10分钟def get_user_stats(user_id):"""获取用户统计信息(耗时操作)"""# 模拟耗时计算importtimetime.sleep(2)return{'total_orders':100,'total_spent':5000.00,'vip_level':3}2. 缓存版本控制
def get_with_version():""" 缓存版本控制 当数据结构变化时,可以通过版本号避免问题"""# 设置缓存,指定版本cache.set('my_key','value',version=2)# 获取缓存,指定版本value_v1=cache.get('my_key',version=1)# Nonevalue_v2=cache.get('my_key',version=2)# 'value'# 删除特定版本cache.delete('my_key',version=2)# 递增版本号cache.incr_version('my_key')八、生产环境配置
# settings/production.pyimportos CACHES={'default':{'BACKEND':'django_redis.cache.RedisCache','LOCATION':os.getenv('REDIS_URL','redis://127.0.0.1:6379/0'),'OPTIONS':{'CLIENT_CLASS':'django_redis.client.DefaultClient','PASSWORD':os.getenv('REDIS_PASSWORD',''),'SOCKET_TIMEOUT':5,# 5秒超时'SOCKET_CONNECT_TIMEOUT':5,# 5秒连接超时'CONNECTION_POOL_KWARGS':{'max_connections':100,# 连接池大小},'COMPRESSOR':'django_redis.compressors.zlib.ZlibCompressor',},'KEY_PREFIX':'myproject_prod',# 生产环境前缀'TIMEOUT':60*30,# 默认30分钟}}九、常见问题解决
1. 缓存穿透(大量查询不存在的数据)
def get_product_safe(product_id):""" 防止缓存穿透:缓存空值""" cache_key=f'product:{product_id}'product=cache.get(cache_key)ifproduct is None:# 查询数据库try: product=Product.objects.get(id=product_id)cache.set(cache_key, product,timeout=3600)except Product.DoesNotExist:# 缓存空值,短时间过期cache.set(cache_key,'NULL',timeout=60)product=Noneelifproduct=='NULL':product=Nonereturnproduct2. 缓存雪崩(大量缓存同时过期)
importrandom def set_with_random_ttl(key, value,base_ttl=3600):""" 设置随机过期时间,避免大量缓存同时过期"""# 添加随机抖动,±300秒ttl=base_ttl + random.randint(-300,300)cache.set(key, value,timeout=ttl)returnttl十、总结
适合使用缓存的场景:
- 频繁读取,很少修改的数据:如商品分类、城市列表
- 计算结果耗时的数据:如统计报表、排行榜
- 会话数据:用户登录状态
- 临时数据:验证码、短信发送频率限制
不适合使用缓存的场景:
- 实时性要求高的数据:如股票价格
- 频繁修改的数据:缓存频繁失效,意义不大
- 敏感数据:除非做好安全措施
简单原则:
- 先让功能跑起来,再考虑优化
- 缓存不是万能的,不要过度使用
- 记得设置合理的过期时间
- 数据更新时,及时清理相关缓存
这个简明教程涵盖了Django中使用Redis缓存的主要方面,从安装配置到实际应用。根据你的项目需求,选择合适的方式使用缓存即可。