每天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 jsonclass CacheManager:def __init__(self, timeout=300):self.timeout = timeoutdef 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 resultreturn wrapperreturn decoratordef 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 datetimeclass 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 Nonedef 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

五、缓存流程图

在这里插入图片描述

六、高级缓存策略

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 valuereturn Nonedef set(self, key, value, timeout=None):# 同时写入本地缓存和Redisself.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, timedeltaclass CacheWarmer:def __init__(self):self.cache = cachedef 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 BaseCommandclass 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 logginglogger = logging.getLogger(__name__)class CacheMonitor:def __init__(self):self.cache = cachedef 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_timeself.log_operation(operation_name, True, elapsed_time)return resultexcept Exception as e:elapsed_time = time.time() - start_timeself.log_operation(operation_name, False, elapsed_time, str(e))raisereturn wrapperreturn decoratordef 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)