Django 从入门到精通：知行博客实战（6.1） · 第 23 / 36 节

事务、并发与业务服务

吴秀峰 3 次阅读 0 条评论 ↓
适用 Django
6.1.2
适用 Python
3.14（兼容 3.12–3.14）
学习难度
入门
最近核验
2026-11-01

前置知识：无需 Django 经验；建议会基本文件操作。Python 与 HTTP 基础可从第 2 章补齐。

学习目标

让“发布文章”和“记录发布事件”保持一致，把重复的业务规则集中到服务函数，理解行锁和 on_commit 的边界。

为什么两次 save 需要一个业务单位

如果文章已公开，但审计事件保存失败，运营记录会与实际状态不一致。atomic 把两步包成事务，失败时一起回滚。已经发布的文章再次保存不应再产生一次发布事件，服务函数需要读取数据库里之前的状态。

select_for_update 在事务内锁住当前文章，避免两个发布请求同时把旧状态看成草稿。它保护状态转换，不是完整的协作编辑冲突解决方案；两个用户先后编辑不同内容仍可能最后写入覆盖前一次内容。需要协作编辑时应增加版本字段或乐观锁。

具体怎么做

  1. 新增 PublicationEvent 模型和迁移，记录文章、操作人和时间。文章删除后保留事件但 post 置空；这里不记录正文等私人内容。
  2. 创建 services.save_post，由新建、编辑、Admin 和后续 API 共同调用。
  3. 已存在记录先检查作者归属，再保存；允许后台超级用户按其运营职责处理。
  4. 提交后再打印发布日志，避免事务回滚却留下“已经发布”的误导事件。生产邮件或任务同样不能早于数据库提交。
  5. 新增两个测试：重复保存不重复记事件、审计失败回滚文章状态。
python manage.py makemigrations blog
python manage.py migrate

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

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

blog/admin.py

from django.contrib import admin
from .models import Category, Post
from .services import save_post

@admin.register(Category)
class CategoryAdmin(admin.ModelAdmin):
    list_display = ("name", "slug")
    search_fields = ("name",)
    prepopulated_fields = {"slug": ("name",)}

@admin.register(Post)
class PostAdmin(admin.ModelAdmin):
    list_display = ("title", "author", "category", "status", "created_at")
    list_filter = ("status", "category")
    search_fields = ("title", "body", "author__username")
    autocomplete_fields = ("category",)
    list_select_related = ("author", "category")
    readonly_fields = ("author", "created_at", "updated_at")

    def get_queryset(self, request):
        queryset = super().get_queryset(request)
        return queryset if request.user.is_superuser else queryset.filter(author=request.user)

    def save_model(self, request, obj, form, change):
        save_post(obj, request.user)

from .models import Comment

@admin.register(Comment)
class CommentAdmin(admin.ModelAdmin):
    list_display = ("id", "post", "author", "approved", "created_at")
    list_filter = ("approved",)
    search_fields = ("body", "author__username")
    list_select_related = ("post", "author")
    readonly_fields = ("post", "author", "body", "created_at")

    def has_add_permission(self, request):
        return False

blog/migrations/0004_publicationevent.py

# Generated by Django 6.1.2 on 2026-11-01 11:13

import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models


class Migration(migrations.Migration):

    dependencies = [
        ('blog', '0003_comment'),
        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
    ]

    operations = [
        migrations.CreateModel(
            name='PublicationEvent',
            fields=[
                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('created_at', models.DateTimeField(auto_now_add=True)),
                ('actor', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to=settings.AUTH_USER_MODEL)),
                ('post', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='publication_events', to='blog.post')),
            ],
        ),
    ]

blog/models.py

from django.conf import settings
from django.db import models
from django.urls import reverse
from .uploads import cover_path

class Category(models.Model):
    name = models.CharField("名称", max_length=40, unique=True)
    slug = models.SlugField("地址标识", unique=True)

    class Meta:
        ordering = ["name", "pk"]

    def __str__(self):
        return self.name

class Post(models.Model):
    class Status(models.TextChoices):
        DRAFT = "draft", "草稿"
        PUBLISHED = "published", "已发布"

    title = models.CharField("标题", max_length=120)
    body = models.TextField("正文")
    author = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="posts")
    category = models.ForeignKey(Category, on_delete=models.PROTECT, related_name="posts")
    status = models.CharField("状态", max_length=12, choices=Status, default=Status.DRAFT)
    cover = models.ImageField("封面", upload_to=cover_path, blank=True)
    created_at = models.DateTimeField("创建时间", auto_now_add=True)
    updated_at = models.DateTimeField("修改时间", auto_now=True)

    class Meta:
        ordering = ["-created_at", "-pk"]
        indexes = [models.Index(fields=["status", "-created_at"], name="post_status_created_idx")]
        constraints = [models.CheckConstraint(condition=models.Q(status__in=["draft", "published"]), name="post_valid_status")]

    def __str__(self):
        return self.title

    def get_absolute_url(self):
        return reverse("blog:detail", args=[self.pk])

class Comment(models.Model):
    post = models.ForeignKey(Post, on_delete=models.CASCADE, related_name="comments")
    author = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    body = models.TextField("评论", max_length=1000)
    approved = models.BooleanField("已审核", default=False)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        ordering = ["-created_at", "-pk"]
        indexes = [models.Index(fields=["author", "-created_at"], name="comment_author_time_idx")]

class PublicationEvent(models.Model):
    post = models.ForeignKey(Post, null=True, on_delete=models.SET_NULL, related_name="publication_events")
    actor = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT)
    created_at = models.DateTimeField(auto_now_add=True)

blog/services.py

import logging
from django.core.exceptions import PermissionDenied
from django.db import transaction
from .models import Post, PublicationEvent

logger = logging.getLogger(__name__)

@transaction.atomic
def save_post(post, actor):
    previous_status = None
    if post.pk:
        current = Post.objects.select_for_update().get(pk=post.pk)
        if current.author_id != actor.pk and not actor.is_superuser:
            raise PermissionDenied
        previous_status = current.status
        post.author_id = current.author_id
    else:
        post.author = actor
    post.save()
    if post.status == Post.Status.PUBLISHED and previous_status != Post.Status.PUBLISHED:
        PublicationEvent.objects.create(post=post, actor=actor)
        transaction.on_commit(lambda: logger.info("post_published post_id=%s actor_id=%s", post.pk, actor.pk))
    return post

blog/test_services.py

from unittest.mock import patch
from django.contrib.auth import get_user_model
from django.test import TestCase
from .models import Post, Category, PublicationEvent
from .services import save_post

class PublicationTests(TestCase):
    def setUp(self):
        self.user = get_user_model().objects.create_user("editor", "editor@example.com")
        category = Category.objects.create(name="实战", slug="practice")
        self.post = Post.objects.create(title="事务实验", body="正文", author=self.user, category=category)

    def test_repeated_save_does_not_repeat_publication_event(self):
        self.post.status = "published"
        save_post(self.post, self.user)
        save_post(self.post, self.user)
        self.assertEqual(PublicationEvent.objects.count(), 1)

    def test_audit_failure_rolls_back_post_change(self):
        self.post.status = "published"
        with patch("blog.services.PublicationEvent.objects.create", side_effect=RuntimeError("模拟失败")):
            with self.assertRaises(RuntimeError):
                save_post(self.post, self.user)
        self.post.refresh_from_db()
        self.assertEqual(self.post.status, "draft")

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 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,
                  "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)

如何确认成功

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

两项测试通过。手工把草稿发布两次，再查询事件：

python manage.py shell -c "from blog.models import PublicationEvent; print(list(PublicationEvent.objects.values('post_id','actor_id')))"

同一次“草稿到发布”只产生一个事件；先撤回草稿再发布，是另一次转换，会产生新事件。后台批量 update 会绕过该服务，当前没有提供这样的发布操作。

练习：模拟事务内第二步异常，检查数据库回滚；思考若 on_commit 后进程立即退出，外部任务是否仍一定送达。需要强一致消息时，应使用数据库 outbox 等可补偿设计，而不是把 on_commit 当作消息可靠性的保证。

常见问题

为何不把所有业务塞入信号？ 显式服务更容易看出执行顺序和错误边界。atomic 能回滚已经发出的邮件吗？ 不能，邮件和数据库不属于同一事务。SQLite 行锁测试通过了吗？ SQLite 会忽略 select_for_update 的锁语义；本章回滚测试能验证原子性，生产并发还要在 MySQL 做多连接测试。

官方参考

把学习连起来

登录后可保存阅读位置、标记章节完成，并在其他设备继续学习。

登录并保存学习进度 →
联系站长

微信：VIPdjango。技术问题建议先在 社区讨论 中交流，方便更多人参与和查阅。

章节讨论 0