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 product

2. 設置合理的超時時間

# 根據數據更新頻率設置超時
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 緩存有哪些層級?

答:

  1. 站點級緩存:Middleware,緩存整個站點
  2. 視圖級緩存:@cache_page,緩存視圖輸出
  3. 模板片段緩存:{% cache %},緩存模板部分
  4. 底層 API:手動控制,最靈活

Q2: cache_page 的原理是什麼?

答:

  1. 根據 URL 和查詢參數生成緩存鍵
  2. 檢查緩存是否存在
  3. 存在則直接返回
  4. 不存在則執行視圖,緩存響應

Q3: 如何避免緩存雪崩?

答:

  1. 設置隨機過期時間
  2. 使用緩存預熱
  3. 多級緩存
  4. 熔斷機制

Q4: Django 緩存的 KEY_PREFIX 有什麼用?

答:

  • 避免不同應用的鍵衝突
  • 方便批量清理
  • 多環境隔離(dev/staging/prod)

🔗 下一篇

在下一篇文章中,我們將深入學習 緩存穿透、擊穿、雪崩,了解這三大緩存問題及其解決方案。

閱讀時間:12 分鐘

0%