Skip to content

极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缓存系统的核心要点:

  1. ✅ 配置合适的缓存后端(Redis、Memcached等)
  2. ✅ 实施页面级、视图级和模板片段缓存
  3. ✅ 设计有效的缓存键命名规范和版本管理
  4. ✅ 建立缓存失效机制确保数据一致性
  5. ✅ 监控缓存性能并进行优化

缓存是提升Web应用性能的重要手段。

下一篇

我们将学习生产环境部署方案。

17.1 生产环境配置 →

目录

返回课程目录

Released under the Apache 2.0 License.