极16章:性能优化
16.2 缓存系统
缓存框架
Django提供了多种缓存后端和策略:
python
# settings.py
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.redis.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/1',
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
}
},
'session': {
'BACKEND': 'django.core.cache.backends.redis.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/2',
}
}
# 使用本地内存缓存(开发环境)
# CACHES = {
# 'default': {
# 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
# 'LOCATION': 'unique-snowflake',
# }
# }
# 使用数据库缓存
# python manage.py createcachetable
# CACHES = {
# 'default': {
# 'BACKEND': 'django.core.cache.backends.db.DatabaseCache',
# 'LOCATION': 'my_cache_table',
# }
# }
# 缓存会话
SESSION_ENGINE = 'django.contrib.sessions.backends.cached_db'
SESSION_CACHE_ALIAS = 'session'
# 缓存中间件配置
MIDDLEWARE = [
'django.middleware.cache.UpdateCacheMiddleware', # 必须是第一个
# ... 其他中间件 ...
'django.middleware.cache.FetchFromCacheMiddleware', # 必须是最后一个
]
CACHE_MIDDLEWARE_ALIAS = 'default'
CACHE_MIDDLEWARE_SECONDS = 600 # 缓存10分钟
CACHE_MIDDLEWARE_KEY_PREFIX = ''页面缓存
页面级别的缓存可以显著提升性能:
python
# views.py
from django.views.decorators.cache import cache_page
from django.utils.decorators import method_decorator
from django.views.generic import ListView, DetailView
# 函数视图页面缓存
@cache_page(60 * 15) # 缓存15分钟
def article_list(request):
articles = Article.objects.select_related('author', 'category').filter(
status='published'
).order_by('-created_at')[:20]
return render(request, 'blog/article_list.html', {'articles': articles})
# 类视图页面缓存
@method_decorator(cache_page(60 * 15), name='dispatch')
class ArticleListView(ListView):
model = Article
template_name = 'blog/article_list.html'
context_object_name = 'articles'
paginate_by = 10
def get_queryset(self):
return Article.objects.select_related('author', 'category').filter(
status='published'
).order_by('-created_at')
# 基于参数的缓存
@cache_page(60 * 15, key_prefix='article_detail')
def article_detail(request, slug):
article = get_object_or_404(
Article.objects.select_related('author', 'category').prefetch_related('tags'),
slug=slug,
status='published'
)
return render(request, 'blog/article_detail.html', {'article': article})
# 条件性页面缓存
from django.views.decorators.vary import vary_on_headers
@cache_page(60 * 15)
@vary_on_headers('User-Agent', 'Cookie')
def responsive_page(request):
# 根据User-Agent或Cookie返回不同内容的页面
return render(request, 'responsive_page.html')
# 自定义缓存键
from django.core.cache import cache
from django.views.decorators.cache import cache_page
from django.utils.cache import get_cache_key
def custom_cache_key(request):
"""生成自定义缓存键"""
key = f"article_detail_{request.user.id if request.user.is_authenticated else 'anonymous'}_{request.GET.get('slug')}"
return key
def article_detail_with_custom_cache(request, slug):
cache_key = custom_cache_key(request)
article = cache.get(cache_key)
if article is None:
article = get_object_or_404(
Article.objects.select_related('author', 'category').prefetch_related('tags'),
slug=slug,
status='published'
)
# 缓存1小时
cache.set(cache_key, article, 60 * 60)
return render(request, 'blog/article_detail.html', {'article': article})视图缓存
视图级别的缓存提供更精细的控制:
python
# views.py
from django.views.decorators.cache import cache_control, never_cache
from django.utils.decorators import method_decorator
from django.core.cache import cache
# 缓存控制装饰器
@cache_control(max_age=3600, public=True) # 缓存1小时,允许公共缓存
def article_list(request):
articles = Article.objects.select_related('author', 'category').filter(
status='published'
).order_by('-created_at')[:20]
return render(request, 'blog/article_list.html', {'articles': articles})
# 禁用缓存
@never_cache
def user_profile(request):
# 用户个人资料不应该被缓存
return render(request, 'accounts/profile.html', {
'profile': request.user.profile
})
# 类视图缓存控制
@method_decorator(cache_control(max_age=3600), name='dispatch')
class ArticleListView(ListView):
# ... 视图实现 ...
# 手动缓存管理
class ArticleDetailView(DetailView):
model = Article
template_name = 'blog/article_detail.html'
context_object_name = 'article'
def get_object(self, queryset=None):
slug = self.kwargs['slug']
cache_key = f'article_{slug}'
# 尝试从缓存获取
article = cache.get(cache_key)
if article is None:
# 缓存未命中,从数据库获取
article = super().get_object(queryset)
# 缓存2小时
cache.set(cache_key, article, 60 * 60 * 2)
else:
# 更新浏览量(即使从缓存获取)
article.view_count += 1
article.save(update_fields=['view_count'])
return article
def dispatch(self, request, *args, **kwargs):
# 检查是否有缓存版本
response = cache.get(f'article_response_{self.kwargs["slug"]}')
if response is not None and not request.user.is_authenticated:
# 只为匿名用户返回缓存响应
return response
response = super().dispatch(request, *args, **kwargs)
# 为匿名用户缓存响应
if not request.user.is_authenticated:
cache.set(
f'article_response_{self.kwargs["slug"]}',
response,
60 * 60 # 缓存1小时
)
return response
# 缓存失效机制
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
@receiver(post_save, sender=Article)
def invalidate_article_cache(sender, instance, **kwargs):
"""文章保存时清除相关缓存"""
cache.delete(f'article_{instance.slug}')
cache.delete_many([
'article_list',
f'user_articles_{instance.author.id}',
f'category_articles_{instance.category.id if instance.category else 0}'
])
@receiver(post_delete, sender=Article)
def invalidate_article_cache_on_delete(sender, instance, **kwargs):
"""文章删除时清除相关缓存"""
cache.delete(f'article_{instance.slug}')
cache.delete('article_list')模板片段缓存
模板级别的缓存可以缓存页面的特定部分:
html
<!-- templates/blog/article_list.html -->
{% load cache %}
<!DOCTYPE html>
<html>
<head>
<title>博客文章列表</title>
</head>
<body>
<!-- 缓存侧边栏1小时 -->
{% cache 3600 sidebar %}
<div class="sidebar">
<h3>热门文章</h3>
{% for article in popular_articles %}
<div class="popular-article">
<a href="{{ article.get_absolute_url }}">{{ article.title }}</a>
</div>
{% endfor %}
</div>
{% endcache %}
<!-- 缓存文章列表30分钟 -->
{% cache 1800 article_list page_obj.number %}
<div class="article-list">
{% for article in articles %}
<article class="article-item">
<h2><a href="{{ article.get_absolute_url }}">{{ article.title }}</a></h2>
<p class="meta">
作者: {{ article.author.username }} |
分类: {{ article.category.name }} |
发布时间: {{ article.created_at|date:"Y-m-d" }}
</p>
<div class="excerpt">{{ article.excerpt|truncatewords:50 }}</div>
</article>
{% endfor %}
</div>
{% endcache %}
<!-- 使用vary_on_cache根据用户状态缓存不同版本 -->
{% load cache %}
{% if user.is_authenticated %}
{% cache 1800 user_welcome user.id %}
<div class="welcome">
<h3>欢迎回来,{{ user.username }}!</h3>
<p>您有 {{ user.unread_notifications.count }} 条未读消息</p>
</div>
{% endcache %}
{% else %}
{% cache 1800 guest_welcome %}
<div class="welcome">
<h3>欢迎访问我们的博客!</h3>
<p><a href="{% url 'accounts:login' %}">登录</a>后可以获得更多功能</p>
</div>
{% endcache %}
{% endif %}
</body>
</html>python
# 在视图中使用模板片段缓存
def article_list(request):
# 为模板缓存准备数据
context = {
'articles': Article.objects.select_related('author', 'category').filter(
status='published'
).order_by('-created_at')[:20],
'popular_articles': cache.get('popular_articles')
}
if context['popular_articles'] is None:
context['popular_articles'] = Article.objects.filter(
status='published',
view_count__gt=1000
).order_by('-view_count')[:5]
# 缓存热门文章1小时
cache.set('popular_articles', context['popular_articles'], 3600)
return render(request, 'blog/article_list.html', context)
# 自定义模板缓存标签
# templatetags/cache_extras.py
from django import template
from django.core.cache import cache
register = template.Library()
@register.simple_tag
def cached_query(queryset, cache_key, timeout=300):
"""缓存查询结果的模板标签"""
result = cache.get(cache_key)
if result is None:
result = list(queryset)
cache.set(cache_key, result, timeout)
return result
# 在模板中使用自定义缓存标签
"""
{% load cache_extras %}
{% cached_query popular_articles "popular_articles" 3600 as cached_popular %}
<div class="sidebar">
<h3>热门文章</h3>
{% for article in cached_popular %}
<div class="popular-article">
<a href="{{ article.get_absolute_url }}">{{ article.title }}</a>
</div>
{% endfor %}
</div>
"""缓存最佳实践
缓存使用的最佳实践:
python
# 缓存键命名规范
class CacheKeys:
"""缓存键命名规范"""
@staticmethod
def article_detail(slug):
return f"article_detail:{slug}"
@staticmethod
def user_profile(user_id):
return f"user_profile:{user_id}"
@staticmethod
def category_articles(category_id, page=1):
return f"category_articles:{category_id}:page:{page}"
@staticmethod
def search_results(query, page=1):
# 清理查询字符串以避免非法字符
clean_query = query.replace(' ', '_').replace(':', '_')
return f"search_results:{clean_query}:page:{page}"
# 缓存版本管理
class CacheVersion:
"""缓存版本管理"""
VERSION = 'v1' # 缓存版本号
@classmethod
def get_key(cls, key):
return f"{cls.VERSION}:{key}"
# 使用示例
def get_article_detail(slug):
cache_key = CacheVersion.get_key(CacheKeys.article_detail(slug))
article = cache.get(cache_key)
if article is None:
article = get_object_or_404(Article, slug=slug)
cache.set(cache_key, article, 3600) # 缓存1小时
return article
# 缓存预热
from django.core.management.base import BaseCommand
class Command(BaseCommand):
"""缓存预热命令"""
def handle(self, *args, **options):
# 预热热门文章
popular_articles = Article.objects.filter(
status='published'
).order_by('-view_count')[:100]
for article in popular_articles:
cache_key = CacheKeys.article_detail(article.slug)
cache.set(cache_key, article, 3600)
# 预热分类页面
categories = Category.objects.all()
for category in categories:
cache_key = f"category_{category.id}_articles"
articles = Article.objects.filter(
category=category, status='published'
).order_by('-created_at')[:20]
cache.set(cache_key, list(articles), 1800)
self.stdout.write(
self.style.SUCCESS('缓存预热完成')
)
# 缓存监控和统计
import logging
from django.core.cache import cache
logger = logging.getLogger(__name__)
class CacheMonitor:
"""缓存监控"""
@staticmethod
def get_cache_stats():
"""获取缓存统计信息"""
try:
# Redis统计信息
from django_redis import get_redis_connection
redis_conn = get_redis_connection("default")
info = redis_conn.info()
return {
'used_memory': info.get('used_memory_human'),
'connected_clients': info.get('connected_clients'),
'total_commands_processed': info.get('total_commands_processed'),
'keyspace_hits': info.get('keyspace_hits'),
'keyspace_misses': info.get('keyspace_misses'),
}
except Exception as e:
logger.error(f"获取缓存统计信息失败: {e}")
return {}
@staticmethod
def cache_hit_rate():
"""计算缓存命中率"""
stats = CacheMonitor.get_cache_stats()
hits = stats.get('keyspace_hits', 0)
misses = stats.get('keyspace_misses', 0)
if hits + misses > 0:
return hits / (hits + misses)
return 0
# 在视图中记录缓存命中情况
def monitored_article_detail(request, slug):
cache_key = CacheKeys.article_detail(slug)
# 记录缓存查询开始
start_time = time.time()
article = cache.get(cache_key)
cache_time = time.time() - start_time
if article is None:
# 缓存未命中
logger.info(f"缓存未命中: {cache_key}, 查询时间: {cache_time:.3f}s")
article = get_object_or_404(Article, slug=slug)
cache.set(cache_key, article, 3600)
else:
# 缓存命中
logger.info(f"缓存命中: {cache_key}, 缓存时间: {cache_time:.3f}s")
return render(request, 'blog/article_detail.html', {'article': article})通过合理使用Django的缓存系统,可以显著提升应用性能,减少数据库负载,并改善用户体验。
小结
Django缓存系统的核心要点:
- ✅ 配置合适的缓存后端(Redis、Memcached等)
- ✅ 实施页面级、视图级和模板片段缓存
- ✅ 设计有效的缓存键命名规范和版本管理
- ✅ 建立缓存失效机制确保数据一致性
- ✅ 监控缓存性能并进行优化
缓存是提升Web应用性能的重要手段。
下一篇
我们将学习生产环境部署方案。