缓存公共数据

学习目标

缓存公开文章数量，区分未命中和零值，理解过期时间、提交后失效与多进程共享问题。

为什么先缓存小而明确的数据

整页缓存可能把用户名、私有草稿或 CSRF 状态带给其他人。先缓存一个不含用户信息的公共统计，能够清楚理解缓存键、值和失效。它是可短暂过期的展示数据，不能用于权限决策、库存或余额。

缓存键加版本后缀，未来改变计算口径时可以使用新键。只把 None 当成未命中，0 是有效统计。文章保存或删除后在事务提交时删除缓存；同时设置 30 秒 TTL，限制多进程局部失效和竞争回填造成的不一致时间。

具体怎么做

1. 新建 statistics.py 和 signals.py，并在 BlogConfig.ready 中注册信号。
2. 使用 LocMemCache，无需额外服务，适合单进程学习。
3. 首页显示公开文章总数，并明确提示可能有短暂延迟。
4. 为缓存的零值添加查询数测试，确保不会在文章为空时每次都重新查库。

本章新增或替换的完整文件

以下路径均相对于练习项目根目录。替换同名文件的全部内容；未列出的文件保持上一章版本。

blog/apps.py

from django.apps import AppConfig

class BlogConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "blog"

    def ready(self):
        from . import signals

blog/signals.py

from django.core.cache import cache
from django.db import transaction
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from .models import Post

@receiver(post_save, sender=Post)
@receiver(post_delete, sender=Post)
def invalidate_public_count(sender, **kwargs):
    transaction.on_commit(lambda: cache.delete("blog:public-post-count:v1"))

blog/statistics.py

from django.core.cache import cache
from .models import Post

def public_post_count():
    key = "blog:public-post-count:v1"
    value = cache.get(key)
    if value is None:
        value = Post.objects.filter(status=Post.Status.PUBLISHED).count()
        cache.set(key, value, timeout=30)
    return value

blog/test_cache.py

from django.core.cache import cache
from django.test import TestCase
from .statistics import public_post_count

class PublicCountTests(TestCase):
    def test_cached_zero_is_not_treated_as_missing(self):
        cache.clear()
        self.addCleanup(cache.clear)
        with self.assertNumQueries(1):
            self.assertEqual(public_post_count(), 0)
        with self.assertNumQueries(0):
            self.assertEqual(public_post_count(), 0)

blog/views.py

from django.shortcuts import render, redirect, get_object_or_404
from django.views.decorators.http import require_GET, require_http_methods
from django.contrib.auth.decorators import login_required
from django.contrib import messages
from .forms import PostForm, CommentForm
from .models import Comment
from .services import save_post
from .statistics import public_post_count
from django.db import transaction
from django.contrib.auth import get_user_model
from django.utils import timezone
from datetime import timedelta
from django.views.decorators.http import require_POST
from django.core.paginator import Paginator
from django.db.models import Q
from .models import Category, Post

@require_GET
def home(request):
    posts = Post.objects.filter(status=Post.Status.PUBLISHED).select_related("author", "category")
    q = request.GET.get("q", "").strip()[:100]
    category = request.GET.get("category", "").strip()[:50]
    if q:
        posts = posts.filter(Q(title__icontains=q) | Q(body__icontains=q))
    if category:
        posts = posts.filter(category__slug=category)
    page_obj = Paginator(posts, 6).get_page(request.GET.get("page"))
    return render(request, "blog/home.html", {"page_obj": page_obj, "q": q,
                  "public_count": public_post_count(), "selected_category": category, "categories": Category.objects.all()})

@require_GET
def detail(request, pk):
    visible = Q(status=Post.Status.PUBLISHED)
    if request.user.is_authenticated:
        visible |= Q(author=request.user)
    post = get_object_or_404(Post.objects.select_related("author", "category").filter(visible), pk=pk)
    comments = post.comments.filter(approved=True).select_related("author")
    page_obj = Paginator(comments, 10).get_page(request.GET.get("page"))
    return render(request, "blog/detail.html", {"post": post, "page_obj": page_obj, "comment_form": CommentForm()})

@require_GET
def about(request):
    return render(request, "blog/about.html")

@require_GET
def old_home(request):
    return redirect("blog:home")

@login_required
@require_GET
def mine(request):
    page_obj = Paginator(Post.objects.filter(author=request.user).select_related("category"), 10).get_page(request.GET.get("page"))
    return render(request, "blog/mine.html", {"page_obj": page_obj})

@login_required
@require_http_methods(["GET", "POST"])
def create(request):
    form = PostForm(request.POST if request.method == "POST" else None, request.FILES or None)
    if request.method == "POST" and form.is_valid():
        post = form.save(commit=False)
        post = save_post(post, request.user)
        messages.success(request, "文章已保存。")
        return redirect(post)
    return render(request, "blog/form.html", {"form": form, "heading": "写文章"})

@login_required
@require_http_methods(["GET", "POST"])
def edit(request, pk):
    post = get_object_or_404(Post, pk=pk, author=request.user)
    form = PostForm(request.POST if request.method == "POST" else None, request.FILES or None, instance=post)
    if request.method == "POST" and form.is_valid():
        post = save_post(form.save(commit=False), request.user)
        messages.success(request, "修改已保存。")
        return redirect(post)
    return render(request, "blog/form.html", {"form": form, "heading": "编辑文章"})

@login_required
@require_http_methods(["GET", "POST"])
def delete(request, pk):
    post = get_object_or_404(Post, pk=pk, author=request.user)
    if request.method == "POST":
        post.delete()
        messages.success(request, "文章已删除。")
        return redirect("blog:mine")
    return render(request, "blog/confirm_delete.html", {"post": post})

@login_required
@require_POST
def comment(request, pk):
    post = get_object_or_404(Post, pk=pk, status=Post.Status.PUBLISHED)
    form = CommentForm(request.POST)
    if not form.is_valid():
        messages.error(request, "评论内容需要 2–1000 个字符，请检查后重新提交。")
        return redirect(post)
    with transaction.atomic():
        # MySQL 上同一用户的并发评论共用这把行锁；SQLite 没有等效的行锁。
        get_user_model().objects.select_for_update().get(pk=request.user.pk)
        recent = Comment.objects.filter(author=request.user, created_at__gte=timezone.now() - timedelta(seconds=30))
        if recent.exists():
            messages.error(request, "评论间隔至少 30 秒，请稍后再试。")
        else:
            Comment.objects.create(post=post, author=request.user, body=form.cleaned_data["body"])
            messages.success(request, "评论已提交，审核后展示。")
    return redirect(post)

config/settings.py

import os
from pathlib import Path
from django.core.exceptions import ImproperlyConfigured

BASE_DIR = Path(__file__).resolve().parent.parent
DEBUG = os.environ.get("DJANGO_DEBUG", "1") == "1"
SECRET_KEY = os.environ.get("DJANGO_SECRET_KEY", "development-only-never-use-this-key-in-production")
if not DEBUG and (SECRET_KEY.startswith("development-") or len(SECRET_KEY) < 50):
    raise ImproperlyConfigured("生产环境必须设置至少 50 字符的独立 DJANGO_SECRET_KEY")
ALLOWED_HOSTS = os.environ.get("DJANGO_ALLOWED_HOSTS", "localhost,127.0.0.1,[::1]").split(",")
INSTALLED_APPS = [
    "django.contrib.admin", "django.contrib.auth", "django.contrib.contenttypes",
    "django.contrib.sessions", "django.contrib.messages", "django.contrib.staticfiles",
    "accounts.apps.AccountsConfig", "blog.apps.BlogConfig",
]
MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.contrib.sessions.middleware.SessionMiddleware",
    "django.middleware.common.CommonMiddleware",
    "django.middleware.csrf.CsrfViewMiddleware",
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "django.contrib.messages.middleware.MessageMiddleware",
    "django.middleware.clickjacking.XFrameOptionsMiddleware",
]
ROOT_URLCONF = "config.urls"
TEMPLATES = [{
    "BACKEND": "django.template.backends.django.DjangoTemplates",
    "DIRS": [BASE_DIR / "templates"], "APP_DIRS": True,
    "OPTIONS": {"context_processors": [
        "django.template.context_processors.request",
        "django.contrib.auth.context_processors.auth",
        "django.contrib.messages.context_processors.messages",
    ]},
}]
WSGI_APPLICATION = "config.wsgi.application"
ASGI_APPLICATION = "config.asgi.application"
DATABASES = {"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": BASE_DIR / "db.sqlite3"}}
AUTH_PASSWORD_VALIDATORS = [
    {"NAME": "django.contrib.auth.password_validation.UserAttributeSimilarityValidator"},
    {"NAME": "django.contrib.auth.password_validation.MinimumLengthValidator"},
    {"NAME": "django.contrib.auth.password_validation.CommonPasswordValidator"},
    {"NAME": "django.contrib.auth.password_validation.NumericPasswordValidator"},
]
LANGUAGE_CODE = "zh-hans"
TIME_ZONE = "Asia/Shanghai"
USE_I18N = True
USE_TZ = True
STATIC_URL = "/static/"
STATICFILES_DIRS = [BASE_DIR / "static"]
STATIC_ROOT = BASE_DIR / "staticfiles"
DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
MAILERS = {"default": {"BACKEND": "django.core.mail.backends.console.EmailBackend"}}

AUTH_USER_MODEL = "accounts.User"

LOGIN_URL = "login"
LOGIN_REDIRECT_URL = "blog:home"
LOGOUT_REDIRECT_URL = "blog:home"

MEDIA_URL = "/media/"
MEDIA_ROOT = BASE_DIR / "media"
DATA_UPLOAD_MAX_MEMORY_SIZE = 3 * 1024 * 1024
FILE_UPLOAD_MAX_MEMORY_SIZE = 2 * 1024 * 1024

# SMTP 可选；没有配置时只打印到开发终端。
if os.environ.get("EMAIL_HOST"):
    MAILERS = {"default": {
        "BACKEND": "django.core.mail.backends.smtp.EmailBackend",
        "OPTIONS": {
            "host": os.environ["EMAIL_HOST"],
            "port": int(os.environ.get("EMAIL_PORT", "587")),
            "username": os.environ.get("EMAIL_HOST_USER", ""),
            "password": os.environ.get("EMAIL_HOST_PASSWORD", ""),
            "use_tls": True,
            "timeout": 10,
        },
    }}
DEFAULT_FROM_EMAIL = os.environ.get("DEFAULT_FROM_EMAIL", "noreply@example.invalid")
PASSWORD_RESET_TIMEOUT = 3600

# 入门采用进程内缓存；多进程间不是共享缓存。
CACHES = {"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache", "LOCATION": "zhixing"}}

templates/blog/home.html

{% extends "base.html" %}
{% block content %}
<h1>学习，实践，分享</h1><p class="muted">共 {{ public_count }} 篇公开文章，统计可能延迟 30 秒。</p>
<form method="get" class="panel" role="search">
<label for="q">搜索文章</label><input id="q" name="q" value="{{ q }}" maxlength="100" placeholder="输入标题或正文关键词">
<label for="category">分类</label><select id="category" name="category"><option value="">全部分类</option>
{% for category in categories %}<option value="{{ category.slug }}"{% if selected_category == category.slug %} selected{% endif %}>{{ category.name }}</option>{% endfor %}</select>
<button type="submit">查找</button> <a href="{% url 'blog:home' %}">清除条件</a></form>
{% for post in page_obj %}<article><h2><a href="{{ post.get_absolute_url }}">{{ post.title }}</a></h2>
<p class="muted">{{ post.author.username }} · {{ post.category.name }} · {{ post.created_at|date:"Y-m-d" }}</p>
<p>{{ post.body|truncatechars:120 }}</p></article>{% empty %}<p>暂时没有公开文章。</p>{% endfor %}
{% include "blog/pagination.html" %}
{% endblock %}

如何确认成功

python manage.py test blog.test_cache --settings=config.test_settings

第一次统计执行一条查询，第二次读取缓存执行零条。使用 runserver 新建一篇已发布文章，刷新首页总数应更新。使用 Gunicorn 多 worker 时，LocMemCache 的各进程不共享内容，页面最多允许约 30 秒统计差异，这是当前实现的已知取舍。

练习：临时将 TTL 改为 3 秒，观察过期后重新查询。为什么不能直接对 mine 页面使用没有用户隔离的公共页面缓存？写出一个会泄露数据的反例。

生产扩展

需要跨进程一致缓存时可选 Django RedisCache，并增加兼容的 redis 客户端依赖和独立 Redis 服务。不要仅修改 BACKEND 就假设部署完成；应配置认证、网络、连接超时、容量、淘汰和故障处理。权限和业务数据仍以数据库为准，缓存不可用应有明确降级策略。

常见问题

信号一定覆盖所有写入吗？ QuerySet.update 和某些批量操作不触发 save 信号，操作结束后要显式失效或依赖有限 TTL。缓存越长越好吗？ 要与业务允许的陈旧时间相匹配。删除缓存就绝无旧数据吗？ 并发请求可能在提交前读到旧值并在删除后回填，严格一致的统计需要更强设计。

官方参考

• 缓存框架
• 应用初始化