每天40分玩转Django：Django缓存视图

Django缓存视图

一、今日学习内容概述

二、缓存配置示例

# 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',
            'PARSER_CLASS': 'redis.connection.HiredisParser',
            'CONNECTION_POOL_CLASS': 'redis.connection.BlockingConnectionPool',
            'CONNECTION_POOL_CLASS_KWARGS': {
                'max_connections': 50,
                'timeout': 20,
            }
        }
    },
    'session': {
        'BACKEND': 'django.core.cache.backends.redis.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/2',
        'TIMEOUT': 86400,  # 1天
    }
}

# 会话引擎配置
SESSION_ENGINE = 'django.contrib.sessions.backends.cache'
SESSION_CACHE_ALIAS = 'session'

三、视图缓存实现

3.1 视图装饰器缓存

# views.py
from django.views.decorators.cache import cache_page
from django.views.decorators.vary import vary_on_cookie, vary_on_headers
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, 'blog/article_list.html', {'articles': articles})

# 基于cookie的缓存
@cache_page(60 * 15)
@vary_on_cookie
def user_articles(request):
    articles = Article.objects.filter(author=request.user)
    return render(request, 'blog/user_articles.html', {'articles': articles})

# 基于请求头的缓存
@cache_page(60 * 15)
@vary_on_headers('User-Agent')
def responsive_view(request):
    # 根据User-Agent返回不同的模板
    user_agent = request.META.get('HTTP_USER_AGENT', '')
    template = 'mobile.html' if 'Mobile' in user_agent else 'desktop.html'
    return render(request, template)

3.2 高级视图缓存

from django.core.cache import cache
from django.conf import settings
from django.utils.cache import get_cache_key
from functools import wraps
import hashlib

def smart_cache_page(timeout=None, cache_key_prefix=None):
    def decorator(view_func):
        @wraps(view_func)
        def wrapper(request, *args, **kwargs):
            # 生成缓存键
            cache_key = generate_cache_key(request, view_func, args, kwargs)
            
            # 尝试从缓存获取响应
            response = cache.get(cache_key)
            if response is not None:
                return response
            
            # 执行视图函数
            response = view_func(request, *args, **kwargs)
            
            # 只缓存成功的响应
            if response.status_code == 200:
                cache.set(cache_key, response, timeout)
            
            return response
        return wrapper
    return decorator

def generate_cache_key(request, view_func, args, kwargs):
    # 基础键
    key_parts = [
        view_func.__name__,
        request.path,
        request.GET.urlencode(),
    ]
    
    # 如果是认证用户，添加用户ID
    if request.user.is_authenticated:
        key_parts.append(str(request.user.id))
    
    # 生成hash
    key = hashlib.md5(''.join(key_parts).encode()).hexdigest()
    return f'view_cache_{key}'

# 使用示例
@smart_cache_page(timeout=300)  # 5分钟缓存
def article_detail(request, article_id):
    article = Article.objects.get(id=article_id)
    return render(request, 'blog/article_detail.html', {'article': article})

3.3 基于会话的缓存管理器

# cache_managers.py
from django.core.cache import caches
from django.conf import settings
import json

class SessionCacheManager:
    def __init__(self, session_key):
        self.session_key = session_key
        self.cache = caches['session']
    
    def _get_user_cache_key(self, key):
        """生成用户特定的缓存键"""
        return f'session_{self.session_key}_{key}'
    
    def get(self, key, default=None):
        """获取缓存数据"""
        cache_key = self._get_user_cache_key(key)
        value = self.cache.get(cache_key)
        return json.loads(value) if value else default
    
    def set(self, key, value, timeout=None):
        """设置缓存数据"""
        cache_key = self._get_user_cache_key(key)
        self.cache.set(cache_key, json.dumps(value), timeout)
    
    def delete(self, key):
        """删除缓存数据"""
        cache_key = self._get_user_cache_key(key)
        self.cache.delete(cache_key)
    
    def clear_all(self):
        """清除该会话的所有缓存"""
        pattern = f'session_{self.session_key}_*'
        self.cache.delete_pattern(pattern)

# 使用中间件自动处理会话缓存
class SessionCacheMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        # 为请求添加缓存管理器
        request.cache_manager = SessionCacheManager(request.session.session_key)
        
        response = self.get_response(request)
        
        return response

四、缓存流程图

五、实际应用示例

5.1 商品列表缓存

# views.py
from django.views.decorators.cache import cache_page
from django.core.cache import cache
from .models import Product

class ProductListView:
    def get_products(self, category=None, page=1):
        # 生成缓存键
        cache_key = f'products_list_{category}_{page}'
        products = cache.get(cache_key)
        
        if products is None:
            # 查询数据库
            queryset = Product.objects.all()
            if category:
                queryset = queryset.filter(category=category)
            
            # 分页
            paginator = Paginator(queryset, 20)
            products = paginator.get_page(page)
            
            # 缓存结果
            cache.set(cache_key, products, 300)  # 5分钟缓存
            
        return products

    @cache_page(60 * 5)
    def list_view(self, request):
        category = request.GET.get('category')
        page = request.GET.get('page', 1)
        products = self.get_products(category, page)
        
        return render(request, 'shop/product_list.html', {
            'products': products
        })

5.2 用户购物车缓存

# cart.py
class CartManager:
    def __init__(self, request):
        self.session_key = request.session.session_key
        self.cache = caches['session']
        
    def get_cart_key(self):
        return f'cart_{self.session_key}'
    
    def get_cart(self):
        return self.cache.get(self.get_cart_key()) or {}
    
    def add_item(self, product_id, quantity=1):
        cart = self.get_cart()
        product_id = str(product_id)
        
        if product_id in cart:
            cart[product_id]['quantity'] += quantity
        else:
            cart[product_id] = {
                'quantity': quantity,
                'added_at': datetime.now().isoformat()
            }
            
        self.cache.set(self.get_cart_key(), cart, 86400)  # 1天
        
    def remove_item(self, product_id):
        cart = self.get_cart()
        product_id = str(product_id)
        
        if product_id in cart:
            del cart[product_id]
            self.cache.set(self.get_cart_key(), cart)
            
    def clear(self):
        self.cache.delete(self.get_cart_key())

六、性能优化建议

1. 缓存键设计

   • 使用有意义的前缀
   • 避免过长的键名
   • 包含版本信息

2. 缓存粒度控制

   • 合理设置过期时间
   • 避免过度缓存
   • 考虑缓存成本

3. 缓存更新策略

   • 主动更新
   • 被动失效
   • 版本控制

4. 缓存监控

   • 命中率统计
   • 内存使用监控
   • 性能分析

七、测试用例

# tests.py
from django.test import TestCase, Client
from django.core.cache import cache
from django.urls import reverse

class ViewCacheTests(TestCase):
    def setUp(self):
        self.client = Client()
        cache.clear()
        
    def test_cached_view(self):
        # 第一次请求
        response1 = self.client.get(reverse('article_list'))
        content1 = response1.content
        
        # 修改数据
        Article.objects.create(title='Test Article')
        
        # 第二次请求应该返回缓存的内容
        response2 = self.client.get(reverse('article_list'))
        content2 = response2.content
        
        self.assertEqual(content1, content2)

八、常见问题和解决方案

1. 缓存穿透

def get_data(self, key):
    # 使用空值标记
    value = cache.get(key)
    if value is None:
        data = expensive_operation()
        if data is None:
            # 缓存空值，避免频繁查询
            cache.set(key, 'NULL', 300)
        else:
            cache.set(key, data)
        return data
    return None if value == 'NULL' else value

2. 缓存雪崩

import random

def set_with_jitter(key, value, timeout):
    """添加随机过期时间，避免同时失效"""
    jitter = random.randint(-60, 60)
    final_timeout = timeout + jitter
    cache.set(key, value, final_timeout)

3. 缓存一致性

def update_data(self, key, value):
    """先更新数据库，再删除缓存"""
    try:
        # 更新数据库
        self.model.objects.filter(pk=key).update(value=value)
        # 删除缓存
        cache.delete(f'data_{key}')
    except Exception as e:
        # 记录错误，可能需要后续处理
        logger.error(f'Failed to update data: {e}')

九、总结

通过本章学习，你应该掌握：

1. Django视图缓存的基本使用
2. 会话缓存的实现方法
3. 缓存性能优化策略
4. 常见缓存问题的解决方案

怎么样今天的内容还满意吗？再次感谢朋友们的观看，关注GZH：凡人的AI工具箱，回复666，送您价值199的AI大礼包。最后，祝您早日实现财务自由，还请给个赞，谢谢！