每天40分玩转Django：Django缓存体系
Django缓存体系一、本日学习内容概述
学习模块重要程度预计学时重要内容缓存基础概念⭐⭐⭐⭐⭐1小时缓存原理、配置方式Django内置缓存后端⭐⭐⭐⭐⭐2小时内存、文件、数据库缓存Redis缓存后端⭐⭐⭐⭐⭐2小时Redis配置、使用方式缓存最佳实践⭐⭐⭐⭐1小时性能优化、策略计划 二、Django缓存配置
2.1 根本配置
# settings.py
# Memcached配置
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.memcached.MemcachedCache',
'LOCATION': '127.0.0.1:11211',
}
}
# Redis配置
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.redis.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/1',
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
'PASSWORD': 'your-password',# 如果有密码
}
}
}
# 文件系统缓存
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.filebased.FileBasedCache',
'LOCATION': '/var/tmp/django_cache',
'TIMEOUT': 300,# 5分钟
'OPTIONS': {
'MAX_ENTRIES': 1000,
'CULL_FREQUENCY': 3,
}
}
}
# 本地内存缓存
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
'LOCATION': 'unique-snowflake',
}
}
三、缓存使用示例
3.1 视图级缓存
# views.py
from django.views.decorators.cache import cache_page
from django.core.cache import cache
from django.shortcuts import render
from .models import Article
# 使用装饰器缓存整个视图
@cache_page(60 * 15)# 缓存15分钟
def article_list(request):
articles = Article.objects.all()
return render(request, 'articles/list.html', {'articles': articles})
# 手动管理缓存
def article_detail(request, pk):
cache_key = f'article_{pk}'
article = cache.get(cache_key)
if article is None:
try:
article = Article.objects.get(pk=pk)
# 缓存文章，有效期1小时
cache.set(cache_key, article, 60 * 60)
except Article.DoesNotExist:
return HttpResponse('文章不存在', status=404)
return render(request, 'articles/detail.html', {'article': article})
3.2 模板片断缓存
<!-- templates/articles/list.html -->
{% load cache %}
<div class="article-list">
{% cache 300 'article_list' %}
{% for article in articles %}
<div class="article-item">
<h2>{{ article.title }}</h2>
<p>{{ article.summary }}</p>
</div>
{% endfor %}
{% endcache %}
</div>
3.3 自定义缓存类
# cache_utils.py
from django.core.cache import cache
from functools import wraps
import hashlib
import json
class CacheManager:
def __init__(self, timeout=300):
self.timeout = timeout
def generate_key(self, *args, **kwargs):
"""生成缓存键"""
key_dict = {
'args': args,
'kwargs': kwargs
}
key_str = json.dumps(key_dict, sort_keys=True)
return hashlib.md5(key_str.encode()).hexdigest()
def cache_decorator(self, prefix=''):
"""缓存装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
cache_key = f"{prefix}_{self.generate_key(*args, **kwargs)}"
result = cache.get(cache_key)
if result is None:
result = func(*args, **kwargs)
cache.set(cache_key, result, self.timeout)
return result
return wrapper
return decorator
def invalidate(self, prefix='', *args, **kwargs):
"""清除缓存"""
cache_key = f"{prefix}_{self.generate_key(*args, **kwargs)}"
cache.delete(cache_key)
# 使用示例
cache_manager = CacheManager(timeout=3600)
@cache_manager.cache_decorator(prefix='get_user_data')
def get_user_data(user_id):
# 耗时的数据库查询
return User.objects.get(id=user_id).get_full_data()
四、Redis缓存实现
4.1 Redis缓存管理器
# redis_cache.py
import json
from django_redis import get_redis_connection
from datetime import datetime
class RedisCacheManager:
def __init__(self, connection='default'):
self.redis = get_redis_connection(connection)
def set_json(self, key, data, timeout=None):
"""存储JSON数据"""
json_data = json.dumps(data)
if timeout:
self.redis.setex(key, timeout, json_data)
else:
self.redis.set(key, json_data)
def get_json(self, key):
"""获取JSON数据"""
data = self.redis.get(key)
if data:
return json.loads(data)
return None
def increment(self, key, amount=1):
"""递增计数器"""
return self.redis.incr(key, amount)
def expire(self, key, timeout):
"""设置过期时间"""
self.redis.expire(key, timeout)
def delete(self, key):
"""删除键"""
self.redis.delete(key)
def clear_prefix(self, prefix):
"""清除指定前缀的所有键"""
keys = self.redis.keys(f"{prefix}:*")
if keys:
self.redis.delete(*keys)
# 使用示例
class ArticleService:
def __init__(self):
self.cache = RedisCacheManager()
def get_article_cache_key(self, article_id):
return f"article:{article_id}"
def get_article(self, article_id):
cache_key = self.get_article_cache_key(article_id)
article_data = self.cache.get_json(cache_key)
if not article_data:
article = Article.objects.get(id=article_id)
article_data = {
'id': article.id,
'title': article.title,
'content': article.content,
'updated_at': article.updated_at.isoformat()
}
self.cache.set_json(cache_key, article_data, timeout=3600)
return article_data
五、缓存流程图
https://i-blog.csdnimg.cn/direct/f80e12a9c30d4a50bf859ed937bafc74.png
六、高级缓存策略
6.1 分层缓存
class LayeredCache:
def __init__(self):
self.local_cache = caches['local']
self.redis_cache = caches['redis']
def get(self, key):
# 先查本地缓存
value = self.local_cache.get(key)
if value is not None:
return value
# 查Redis缓存
value = self.redis_cache.get(key)
if value is not None:
# 写入本地缓存
self.local_cache.set(key, value, timeout=60)
return value
return None
def set(self, key, value, timeout=None):
# 同时写入本地缓存和Redis
self.local_cache.set(key, value, timeout=min(timeout, 60) if timeout else 60)
self.redis_cache.set(key, value, timeout=timeout)
6.2 缓存预热
# cache_warmer.py
from django.core.cache import cache
from django.db.models import Q
from datetime import datetime, timedelta
class CacheWarmer:
def __init__(self):
self.cache = cache
def warm_article_cache(self):
"""预热热门文章缓存"""
# 获取最近一周的热门文章
recent_date = datetime.now() - timedelta(days=7)
popular_articles = Article.objects.filter(
Q(created_at__gte=recent_date) |
Q(view_count__gte=1000)
).select_related('author')
for article in popular_articles:
cache_key = f'article_{article.id}'
self.cache.set(cache_key, article, timeout=3600)
return len(popular_articles)
# 在Django管理命令中使用
from django.core.management.base import BaseCommand
class Command(BaseCommand):
help = '预热缓存'
def handle(self, *args, **kwargs):
warmer = CacheWarmer()
count = warmer.warm_article_cache()
self.stdout.write(
self.style.SUCCESS(f'Successfully warmed cache for {count} articles')
)
七、缓存监控和管理
# cache_monitor.py
from django.core.cache import cache
import time
import logging
logger = logging.getLogger(__name__)
class CacheMonitor:
def __init__(self):
self.cache = cache
def monitor_operation(self, operation_name):
"""监控缓存操作的装饰器"""
def decorator(func):
def wrapper(*args, **kwargs):
start_time = time.time()
try:
result = func(*args, **kwargs)
elapsed_time = time.time() - start_time
self.log_operation(operation_name, True, elapsed_time)
return result
except Exception as e:
elapsed_time = time.time() - start_time
self.log_operation(operation_name, False, elapsed_time, str(e))
raise
return wrapper
return decorator
def log_operation(self, operation, success, elapsed_time, error=None):
"""记录缓存操作日志"""
log_data = {
'operation': operation,
'success': success,
'elapsed_time': elapsed_time,
'error': error
}
if success:
logger.info(f'Cache operation: {operation}', extra=log_data)
else:
logger.error(f'Cache operation failed: {operation}', extra=log_data)
八、性能优化发起
[*]缓存键计划
[*]使用故意义的前缀
[*]制止过长的键名
[*]思量版本号机制
[*]缓存粒度
[*]公道分别缓存单元
[*]制止过大的缓存对象
[*]思量部门字段缓存
[*]缓存策略
[*]设置合适的过期时间
[*]实现缓存预热机制
[*]使用分层缓存提高命中率
[*]缓存清理
[*]及时清理过期缓存
[*]实现缓存更新机制
[*]制止缓存穿透
九、总结
本日我们学习了：
[*]Django缓存体系的根本配置
[*]多种缓存后端的使用方法
[*]自定义缓存管理器的实现
[*]缓存性能优化策略
怎么样本日的内容还满意吗？再次感谢朋友们的观看，关注GZH：凡人的AI工具箱，回复666，送您代价199的AI大礼包。末了，祝您早日实现财政自由，还请给个赞，谢谢！
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]