Django 面試準備 09-2:Django 緩存框架
掌握 Django 內建緩存系統的各種使用方式
目錄
09-2. Django 緩存框架
📌 Django 緩存層級
Django 提供了多層次的緩存機制:
1. 站點級緩存(Middleware)
↓
2. 視圖級緩存(Decorator)
↓
3. 模板片段緩存(Template Tag)
↓
4. 底層緩存 API(Manual)🔧 緩存後端配置
1. Redis(推薦)
# settings.py
CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/1',
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
'CONNECTION_POOL_KWARGS': {
'max_connections': 50,
'retry_on_timeout': True,
},
'PASSWORD': 'your-password', # 如果有密碼
'SOCKET_CONNECT_TIMEOUT': 5,
'SOCKET_TIMEOUT': 5,
},
'KEY_PREFIX': 'myapp', # 鍵前綴
'VERSION': 1, # 版本號
'TIMEOUT': 300, # 默認超時(秒)
}
}2. Memcached
# settings.py
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.memcached.PyMemcacheCache',
'LOCATION': [
'127.0.0.1:11211',
'127.0.0.1:11212',
],
'OPTIONS': {
'no_delay': True,
'ignore_exc': True,
'max_pool_size': 4,
'use_pooling': True,
},
'TIMEOUT': 300,
}
}3. 本地內存(開發)
# settings.py
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
'LOCATION': 'unique-snowflake',
'TIMEOUT': 300,
'OPTIONS': {
'MAX_ENTRIES': 1000
}
}
}4. 文件系統(不推薦生產環境)
# settings.py
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.filebased.FileBasedCache',
'LOCATION': '/var/tmp/django_cache',
'TIMEOUT': 300,
}
}🎯 站點級緩存
緩存整個網站的所有頁面。
配置
# settings.py
MIDDLEWARE = [
'django.middleware.cache.UpdateCacheMiddleware', # 最前面
'django.middleware.common.CommonMiddleware',
# ... 其他 middleware
'django.middleware.cache.FetchFromCacheMiddleware', # 最後面
]
# 緩存設置
CACHE_MIDDLEWARE_ALIAS = 'default'
CACHE_MIDDLEWARE_SECONDS = 600 # 10 分鐘
CACHE_MIDDLEWARE_KEY_PREFIX = 'mysite'⚠️ 注意事項
站點級緩存會緩存所有頁面,包括:
- ❌ 用戶個性化內容
- ❌ 動態數據
- ❌ 需要登錄的頁面
適用場景:
- ✅ 靜態內容網站
- ✅ 所有用戶看到相同內容
- ✅ 更新頻率低的網站
🎯 視圖級緩存
緩存特定視圖的輸出。
基本用法
from django.views.decorators.cache import cache_page
# 緩存 5 分鐘
@cache_page(60 * 5)
def product_list(request):
products = Product.objects.all()
return render(request, 'products/list.html', {
'products': products
})URL 配置中使用
# urls.py
from django.views.decorators.cache import cache_page
urlpatterns = [
path('products/', cache_page(60 * 5)(product_list)),
path('about/', cache_page(60 * 15)(about_view)),
]類視圖緩存
from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_page
from django.views.generic import ListView
# 方法 1:裝飾 dispatch
class ProductListView(ListView):
model = Product
template_name = 'products/list.html'
@method_decorator(cache_page(60 * 5))
def dispatch(self, *args, **kwargs):
return super().dispatch(*args, **kwargs)
# 方法 2:在 URL 配置中
urlpatterns = [
path('products/', cache_page(60 * 5)(ProductListView.as_view())),
]條件緩存
from django.views.decorators.cache import cache_page
def product_list(request):
# 只對匿名用戶緩存
if request.user.is_authenticated:
products = Product.objects.all()
return render(request, 'products/list.html', {
'products': products
})
# 匿名用戶使用緩存
@cache_page(60 * 5)
def _cached_view(request):
products = Product.objects.all()
return render(request, 'products/list.html', {
'products': products
})
return _cached_view(request)🎯 模板片段緩存
緩存模板中的部分內容。
基本用法
{% load cache %}
{% cache 500 sidebar %}
<div class="sidebar">
{% for item in menu_items %}
<a href="{{ item.url }}">{{ item.name }}</a>
{% endfor %}
</div>
{% endcache %}帶變量的緩存
{% load cache %}
<!-- 為每個用戶緩存不同的內容 -->
{% cache 500 sidebar request.user.username %}
<div class="sidebar">
<h3>Welcome, {{ request.user.username }}!</h3>
<!-- 用戶專屬內容 -->
</div>
{% endcache %}
<!-- 為每個產品分類緩存不同的內容 -->
{% cache 300 product_list category.id %}
<div class="products">
{% for product in products %}
<div class="product">{{ product.name }}</div>
{% endfor %}
</div>
{% endcache %}指定緩存後端
{% load cache %}
<!-- 使用特定的緩存後端 -->
{% cache 500 sidebar using="redis" %}
<!-- 內容 -->
{% endcache %}🎯 底層緩存 API
最靈活的緩存方式,完全手動控制。
基本操作
from django.core.cache import cache
# 設置緩存
cache.set('my_key', 'my_value', timeout=300) # 5 分鐘
# 獲取緩存
value = cache.get('my_key')
if value is None:
value = expensive_computation()
cache.set('my_key', value, timeout=300)
# 獲取或設置(更簡潔)
value = cache.get_or_set('my_key', expensive_computation, timeout=300)
# 刪除緩存
cache.delete('my_key')
# 清空所有緩存
cache.clear()批量操作
# 批量設置
cache.set_many({
'key1': 'value1',
'key2': 'value2',
'key3': 'value3'
}, timeout=300)
# 批量獲取
values = cache.get_many(['key1', 'key2', 'key3'])
# {'key1': 'value1', 'key2': 'value2', 'key3': 'value3'}
# 批量刪除
cache.delete_many(['key1', 'key2', 'key3'])原子操作
# 增加值(原子操作)
cache.set('counter', 0)
cache.incr('counter') # 1
cache.incr('counter', delta=5) # 6
# 減少值
cache.decr('counter') # 5
cache.decr('counter', delta=2) # 3永不過期
# timeout=None 表示永不過期
cache.set('permanent_key', 'permanent_value', timeout=None)🎯 實戰案例
案例 1:緩存數據庫查詢
from django.core.cache import cache
from django.db.models import Count
def get_hot_products():
"""獲取熱門商品(緩存 10 分鐘)"""
cache_key = 'hot_products'
products = cache.get(cache_key)
if products is None:
# 複雜查詢
products = Product.objects.annotate(
order_count=Count('order_items')
).filter(
order_count__gte=10
).order_by('-order_count')[:10]
# 轉換為列表以便序列化
products = list(products.values(
'id', 'name', 'price', 'order_count'
))
# 緩存 10 分鐘
cache.set(cache_key, products, timeout=600)
return products案例 2:緩存函數計算結果
import hashlib
from django.core.cache import cache
def cache_function(timeout=300):
"""裝飾器:緩存函數結果"""
def decorator(func):
def wrapper(*args, **kwargs):
# 生成緩存鍵
key_data = f"{func.__name__}:{args}:{kwargs}"
cache_key = hashlib.md5(key_data.encode()).hexdigest()
# 嘗試從緩存獲取
result = cache.get(cache_key)
if result is not None:
return result
# 執行函數
result = func(*args, **kwargs)
# 緩存結果
cache.set(cache_key, result, timeout=timeout)
return result
return wrapper
return decorator
# 使用
@cache_function(timeout=600)
def calculate_statistics(user_id, start_date, end_date):
"""計算用戶統計數據(耗時操作)"""
# 複雜計算...
return statistics案例 3:緩存用戶會話數據
from django.core.cache import cache
class UserProfile:
@staticmethod
def get_profile(user_id):
"""獲取用戶資料(優先從緩存)"""
cache_key = f'user_profile:{user_id}'
profile = cache.get(cache_key)
if profile is None:
try:
user = User.objects.select_related('profile').get(id=user_id)
profile = {
'id': user.id,
'username': user.username,
'email': user.email,
'avatar': user.profile.avatar.url if user.profile else None,
}
# 緩存 30 分鐘
cache.set(cache_key, profile, timeout=1800)
except User.DoesNotExist:
return None
return profile
@staticmethod
def update_profile(user_id, data):
"""更新用戶資料並清除緩存"""
user = User.objects.get(id=user_id)
# 更新數據庫...
# 清除緩存
cache_key = f'user_profile:{user_id}'
cache.delete(cache_key)案例 4:緩存分頁數據
from django.core.cache import cache
from django.core.paginator import Paginator
def get_product_page(page_number=1, per_page=20):
"""獲取商品分頁(緩存每頁)"""
cache_key = f'products:page:{page_number}:per_page:{per_page}'
page_data = cache.get(cache_key)
if page_data is None:
products = Product.objects.filter(is_active=True).order_by('-created_at')
paginator = Paginator(products, per_page)
page = paginator.get_page(page_number)
page_data = {
'products': list(page.object_list.values(
'id', 'name', 'price', 'image'
)),
'has_next': page.has_next(),
'has_previous': page.has_previous(),
'total_pages': paginator.num_pages,
}
# 緩存 5 分鐘
cache.set(cache_key, page_data, timeout=300)
return page_data案例 5:緩存 API 響應
from django.core.cache import cache
from rest_framework.decorators import api_view
from rest_framework.response import Response
@api_view(['GET'])
def api_statistics(request):
"""API:統計數據(緩存 15 分鐘)"""
cache_key = 'api:statistics'
data = cache.get(cache_key)
if data is None:
data = {
'total_users': User.objects.count(),
'total_orders': Order.objects.count(),
'total_revenue': Order.objects.aggregate(
total=Sum('total_amount')
)['total'],
'timestamp': timezone.now().isoformat(),
}
# 緩存 15 分鐘
cache.set(cache_key, data, timeout=900)
return Response(data)🔍 緩存鍵管理
最佳實踐
# ❌ 不好的鍵命名
cache.set('data', value)
cache.set('user', user_data)
# ✅ 好的鍵命名
cache.set('product:list:featured', products)
cache.set(f'user:profile:{user_id}', user_data)
cache.set(f'order:detail:{order_id}', order_data)
# 使用類封裝
class CacheKeys:
@staticmethod
def user_profile(user_id):
return f'user:profile:{user_id}'
@staticmethod
def product_list(category_id, page):
return f'product:list:category:{category_id}:page:{page}'
@staticmethod
def order_detail(order_id):
return f'order:detail:{order_id}'
# 使用
cache.set(CacheKeys.user_profile(123), user_data)
cache.get(CacheKeys.product_list(5, 1))🛠️ 緩存失效策略
1. 主動失效
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
@receiver(post_save, sender=Product)
def invalidate_product_cache(sender, instance, **kwargs):
"""商品更新時清除相關緩存"""
# 清除商品詳情緩存
cache.delete(f'product:detail:{instance.id}')
# 清除商品列表緩存
cache.delete('product:list:all')
cache.delete(f'product:list:category:{instance.category_id}')
@receiver(post_delete, sender=Product)
def invalidate_product_list_cache(sender, instance, **kwargs):
"""商品刪除時清除列表緩存"""
cache.delete('product:list:all')
cache.delete(f'product:list:category:{instance.category_id}')2. 模式匹配刪除(Redis)
from django_redis import get_redis_connection
def clear_cache_pattern(pattern):
"""刪除匹配模式的所有緩存鍵"""
redis_conn = get_redis_connection("default")
keys = redis_conn.keys(pattern)
if keys:
redis_conn.delete(*keys)
# 使用
clear_cache_pattern('product:*') # 清除所有商品相關緩存
clear_cache_pattern('user:profile:*') # 清除所有用戶資料緩存3. 版本控制
from django.core.cache import cache
# 使用版本號
cache.set('data', value, version=1)
cache.get('data', version=1)
# 更新時增加版本號
cache.set('data', new_value, version=2)💡 最佳實踐
1. 避免緩存穿透
def get_product(product_id):
"""獲取商品(防止緩存穿透)"""
cache_key = f'product:{product_id}'
product = cache.get(cache_key)
if product is None:
try:
product = Product.objects.get(id=product_id)
cache.set(cache_key, product, timeout=300)
except Product.DoesNotExist:
# 緩存空值,防止重複查詢
cache.set(cache_key, 'NOT_FOUND', timeout=60)
return None
if product == 'NOT_FOUND':
return None
return product2. 設置合理的超時時間
# 根據數據更新頻率設置超時
CACHE_TIMEOUTS = {
'hot_data': 60, # 熱點數據:1 分鐘
'normal_data': 300, # 普通數據:5 分鐘
'static_data': 3600, # 靜態數據:1 小時
'rarely_changed': 86400, # 很少變化:1 天
}
cache.set('hot_products', products, timeout=CACHE_TIMEOUTS['hot_data'])
cache.set('site_config', config, timeout=CACHE_TIMEOUTS['static_data'])3. 監控緩存命中率
from django.core.cache import cache
class CacheMetrics:
@staticmethod
def track_hit(key):
"""記錄緩存命中"""
cache.incr(f'metrics:hit:{key}', 1)
@staticmethod
def track_miss(key):
"""記錄緩存未命中"""
cache.incr(f'metrics:miss:{key}', 1)
@staticmethod
def get_hit_rate(key):
"""計算命中率"""
hits = cache.get(f'metrics:hit:{key}', 0)
misses = cache.get(f'metrics:miss:{key}', 0)
total = hits + misses
return (hits / total * 100) if total > 0 else 0
# 使用
def get_cached_data(key):
data = cache.get(key)
if data is not None:
CacheMetrics.track_hit(key)
else:
CacheMetrics.track_miss(key)
data = fetch_from_db()
cache.set(key, data)
return data💡 面試要點
Q1: Django 緩存有哪些層級?
答:
- 站點級緩存:Middleware,緩存整個站點
- 視圖級緩存:@cache_page,緩存視圖輸出
- 模板片段緩存:{% cache %},緩存模板部分
- 底層 API:手動控制,最靈活
Q2: cache_page 的原理是什麼?
答:
- 根據 URL 和查詢參數生成緩存鍵
- 檢查緩存是否存在
- 存在則直接返回
- 不存在則執行視圖,緩存響應
Q3: 如何避免緩存雪崩?
答:
- 設置隨機過期時間
- 使用緩存預熱
- 多級緩存
- 熔斷機制
Q4: Django 緩存的 KEY_PREFIX 有什麼用?
答:
- 避免不同應用的鍵衝突
- 方便批量清理
- 多環境隔離(dev/staging/prod)
🔗 下一篇
在下一篇文章中,我們將深入學習 緩存穿透、擊穿、雪崩,了解這三大緩存問題及其解決方案。
閱讀時間:12 分鐘