CodeRoad Map 程序员的学习成长路线图。内容免费开放，登录后可同步路线图与课时进度。
由 程序员柏风 独立维护 · 关于作者
学习 更多
DRF 与异步任务 REST、序列化与 Celery。
DRF 与异步任务
这一课你将能做到
用 Django REST Framework 构建符合 RESTful 规范的 API，覆盖序列化、视图集、路由、权限
用 Celery + Redis 处理耗时任务（发邮件、生成报表、图片处理），不阻塞 HTTP 响应
理解什么时候用 Django 模板渲染（全栈）、什么时候用 DRF（前后端分离）
为什么要学
现代 Web 应用大部分走前后端分离：前端 React/Vue/小程序，后端只暴露 JSON API。Django 原生视图写 API 要手动序列化、手动处理 Content-Type，重复代码多。DRF（Django REST Framework）把这些都封装好，让 Django 变成专业的 API 框架。
和本站（codeforge）的对比：本站用 Spring Boot + Spring Web 写 REST API，用 @RestController + @GetMapping。DRF 用 ViewSet + Router，思路完全一致：声明式路由、自动序列化、权限拦截。Spring 用 Jackson 序列化，DRF 用 Serializer。
异步任务：发邮件、生成报表、图片压缩这类操作可能要几秒到几分钟，不能让 HTTP 请求一直等。Celery 是 Python 生态最成熟的分布式任务队列，把任务丢到 Redis/RabbitMQ，由独立 worker 进程异步执行。
先建立图像
前端 React DRF Celery Worker
│ │ │ │
│ POST /api/articles/ │ │ │
├──────────────────────>│ │ │
│ │ serializer.save() │ │
│ │ ──> 任务落库 │ │
│ │ │ │
│ │ send_email.delay() │ │
│ ├─────────────────────>│ │
│ 201 Created │ │ ──> 入队 ──> 取出 ──>│
│<──────────────────────│ │ │
│ (立即返回，不等邮件) │ │ │ 发邮件
跟着做
步骤 1：安装和配置 DRF
pip install djangorestframework
pip install django-filter # 可选：强大的过滤
# settings.py
INSTALLED_APPS = [
...
"rest_framework",
"django_filters",
"blog",
]
REST_FRAMEWORK = {
"DEFAULT_PAGINATION_CLASS": "rest_framework.pagination.PageNumberPagination",
"PAGE_SIZE": 20,
"DEFAULT_FILTER_BACKENDS": [
"django_filters.rest_framework.DjangoFilterBackend",
"rest_framework.filters.SearchFilter",
"rest_framework.filters.OrderingFilter",
],
"DEFAULT_AUTHENTICATION_CLASSES": [
"rest_framework.authentication.SessionAuthentication",
"rest_framework.authentication.TokenAuthentication",
],
"DEFAULT_PERMISSION_CLASSES": [
"rest_framework.permissions.IsAuthenticatedOrReadOnly",
],
}
步骤 2：序列化器（Serializer） from rest_framework import serializers
from .models import Article, Category, Tag
from django.contrib.auth.models import User
class CategorySerializer(serializers.ModelSerializer):
article_count = serializers.IntegerField(read_only=True)
class Meta:
model = Category
fields = ["id", "name", "slug", "article_count"]
class TagSerializer(serializers.ModelSerializer):
class Meta:
model = Tag
fields = ["id", "name"]
class AuthorSerializer(serializers.ModelSerializer):
class Meta:
model = User
fields = ["id", "username", "first_name"]
class ArticleListSerializer(serializers.ModelSerializer):
"""列表页用轻量序列化器"""
author = AuthorSerializer(read_only=True)
category = CategorySerializer(read_only=True)
tags = TagSerializer(many=True, read_only=True)
class Meta:
model = Article
fields = ["id", "title", "slug", "author", "category", "tags",
"views", "status", "created_at"]
class ArticleDetailSerializer(serializers.ModelSerializer):
"""详情页用完整序列化器"""
author = AuthorSerializer(read_only=True)
category = CategorySerializer(read_only=True)
tags = TagSerializer(many=True, read_only=True)
category_id = serializers.PrimaryKeyRelatedField(
queryset=Category.objects.all(), source="category", write_only=True
)
tag_ids = serializers.PrimaryKeyRelatedField(
queryset=Tag.objects.all(), source="tags", many=True, write_only=True
)
class Meta:
model = Article
fields = ["id", "title", "slug", "content", "author", "category",
"category_id", "tags", "tag_ids", "views", "status",
"created_at", "updated_at"]
read_only_fields = ["views", "created_at", "updated_at"]
def validate_slug(self, value):
qs = Article.objects.filter(slug=value)
if self.instance:
qs = qs.exclude(pk=self.instance.pk)
if qs.exists():
raise serializers.ValidationError("该 slug 已被使用")
return value
步骤 3：ViewSet 和 Router from rest_framework import viewsets, filters, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated, IsAuthenticatedOrReadOnly
from django_filters.rest_framework import DjangoFilterBackend
from .models import Article, Category
from .serializers import (
ArticleListSerializer, ArticleDetailSerializer, CategorySerializer
)
class CategoryViewSet(viewsets.ModelViewSet):
queryset = Category.objects.all()
serializer_class = CategorySerializer
lookup_field = "slug" # 用 slug 而不是 id 作为 URL 标识
class ArticleViewSet(viewsets.ModelViewSet):
queryset = Article.objects.select_related("author", "category").prefetch_related("tags")
permission_classes = [IsAuthenticatedOrReadOnly]
filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
filterset_fields = ["status", "category__slug", "author__username"]
search_fields = ["title", "content"]
ordering_fields = ["created_at", "views"]
ordering = ["-created_at"]
lookup_field = "slug"
def get_serializer_class(self):
if self.action == "list":
return ArticleListSerializer
return ArticleDetailSerializer
def perform_create(self, serializer):
serializer.save(author=self.request.user)
@action(detail=True, methods=["post"], permission_classes=[IsAuthenticated])
def like(self, request, slug=None):
"""自定义动作：POST /api/articles/{slug}/like/"""
article = self.get_object()
article.views += 1
article.save(update_fields=["views"])
return Response({"views": article.views})
@action(detail=False, methods=["get"])
def hot(self, request):
"""自定义集合动作：GET /api/articles/hot/"""
hot = self.get_queryset().filter(views__gt=100)[:10]
serializer = self.get_serializer(hot, many=True)
return Response(serializer.data)
# mysite/urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from blog.views import ArticleViewSet, CategoryViewSet
router = DefaultRouter()
router.register(r"articles", ArticleViewSet, basename="article")
router.register(r"categories", CategoryViewSet, basename="category")
urlpatterns = [
path("admin/", admin.site.urls),
path("api/", include(router.urls)),
path("api-auth/", include("rest_framework.urls")), # 浏览器登录
]
GET /api/articles/ → list
POST /api/articles/ → create
GET /api/articles/{slug}/ → retrieve
PUT /api/articles/{slug}/ → update
PATCH /api/articles/{slug}/ → partial_update
DELETE /api/articles/{slug}/ → destroy
POST /api/articles/{slug}/like/ → like（自定义）
GET /api/articles/hot/ → hot（自定义）
步骤 4：Token 认证（适合 SPA / 移动端） # 在 settings.py 的 INSTALLED_APPS 加
"rest_framework.authtoken",
# 迁移
python manage.py migrate
# 视图：获取 token
from rest_framework.authtoken.views import obtain_auth_token
# urls.py
path("api/token/", obtain_auth_token),
# 1. 用账号密码换 token
curl -X POST http://localhost:8000/api/token/ \
-d "username=alice&password=pass123"
# {"token": "abc123..."}
# 2. 后续请求带 token
curl -H "Authorization: Token abc123..." http://localhost:8000/api/articles/
步骤 5：Celery 异步任务 # mysite/celery.py
import os
from celery import Celery
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
app = Celery("mysite")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()
# mysite/__init__.py
from .celery import app as celery_app
__all__ = ("celery_app",)
# settings.py
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/1"
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_TIMEZONE = "Asia/Shanghai"
# blog/tasks.py
from celery import shared_task
from django.core.mail import send_mail
from .models import Article
@shared_task(bind=True, max_retries=3)
def send_article_notification(self, article_id):
try:
article = Article.objects.get(id=article_id)
send_mail(
subject=f"新文章发布：{article.title}",
message=f"作者 {article.author.username} 发布了新文章",
from_email="noreply@example.com",
recipient_list=["admin@example.com"],
)
except Exception as exc:
raise self.retry(exc=exc, countdown=60) # 60 秒后重试
@shared_task
def daily_report():
from django.db.models import Count
from django.utils import timezone
today = timezone.now().date()
count = Article.objects.filter(created_at__date=today).count()
# 生成报表、发邮件、写缓存 ...
return f"今日新增文章：{count}"
from .tasks import send_article_notification
class ArticleViewSet(viewsets.ModelViewSet):
def perform_create(self, serializer):
article = serializer.save(author=self.request.user)
# 异步发送通知，不阻塞响应
send_article_notification.delay(article.id)
# 终端 1：启动 Redis（本地需先装 Redis）
redis-server
# 终端 2：启动 Celery worker
celery -A mysite worker --loglevel=info
# 终端 3（可选）：定时任务调度器
celery -A mysite beat --loglevel=info
# settings.py
from celery.schedules import crontab
CELERY_BEAT_SCHEDULE = {
"daily-report": {
"task": "blog.tasks.daily_report",
"schedule": crontab(hour=8, minute=0), # 每天 8:00
},
}
讲清楚要点
Serializer 是 DRF 的灵魂 ：类似 Spring 的 DTO + Jackson，负责「模型 → JSON」和「JSON → 模型」双向转换。可以定义多个 serializer 用于不同场景（列表/详情）。ViewSet + Router 消灭 90% 的路由代码 ：一个 ViewSet 自动产出 CRUD 五个端点。@action 装饰器加自定义动作（如 like、hot）。权限分两层 ：permission_classes 控制「能不能访问这个视图」（如 IsAuthenticated）；get_queryset() 控制「能看到哪些数据」（如只看自己的）。两层都要做。perform_create 是写入额外字段的钩子request.user。Celery 解决「HTTP 请求不能慢」的问题 ：把耗时操作（邮件、报表、第三方 API 调用）丢给后台 worker，HTTP 立即返回。Redis 既当 broker（任务队列）又当 backend（结果存储）。Session 认证 vs Token 认证 ：Session 适合同站浏览器（带 cookie），Token 适合 SPA/移动端（无状态）。DRF 默认同时支持。
你可能会懵的地方
DRF 和 Django 原生视图怎么选？
返回 HTML → Django 原生（TemplateView、ListView）。返回 JSON → DRF。混用时，页面渲染走原生，AJAX 接口走 DRF。
read_only_fields 和 write_only 的区别？read_only_fields：序列化时输出，反序列化时忽略（如 created_at）。write_only：反序列化时接受，序列化时不输出（如密码、category_id 这种 ID 字段）。
Celery 任务失败了怎么办？
用 bind=True + self.retry() 自动重试。还可以配置 task_annotations 设置不同任务的重试策略。彻底失败的任务进死信队列，需要监控。
delay() 和 apply_async() 区别？delay(args) 是 apply_async(args) 的简写。apply_async 支持更多参数：countdown（延迟执行）、eta（指定时间）、expires（过期时间）、queue（指定队列）。
DRF 分页怎么定制？
DEFAULT_PAGINATION_CLASS 可换成 LimitOffsetPagination（?limit=10&offset=20）或 CursorPagination（高性能、防跳过）。也可以在 ViewSet 上单独覆盖 pagination_class。
错法对照 错法 问题 正解 序列化器里嵌套关系不指定 read_only POST 时要求传完整嵌套对象 嵌套 serializer 加 read_only=True，写入用 xxx_id 字段 Celery 任务函数传 ORM 对象 序列化失败 / 对象过时 传 ID，任务里重新查询 同步视图里调用耗时 API 用户等几秒，体验差 立即返回，Celery 异步处理 @action 忘记 detail=True/False路由 404 单对象用 detail=True，集合用 detail=False queryset 属性不加 select_related列表接口 N+1 get_queryset() 里加 prefetch全局 IsAuthenticated 但登录接口也要登录 鸡生蛋问题 登录/注册视图用 AllowAny
卡点急救
DRF 浏览器访问 API 返回 403 ：Session 认证要求 CSRF。浏览器打开 /api-auth/login/ 登录后再访问。Celery 任务不执行 ：检查 Redis 是否启动，worker 是否在跑，任务是否被 @shared_task 装饰。celery -A mysite inspect active 查看正在执行的任务。AssertionError: 'basename' argument not specifiedqueryset 没设置，无法推断 basename。手动加 basename="article"。
面试常问
DRF 的 Serializer 和 Django Form 有什么区别？
Form 主要处理 HTML 表单，输出带 CSRF 的 HTML。Serializer 处理 JSON/XML 等数据格式，不渲染 HTML，支持嵌套序列化和复杂校验。两者 API 相似（is_valid()、validated_data/cleaned_data），但用途不同。
Celery 的 broker 和 backend 分别是什么？
broker 是任务队列（Redis/RabbitMQ），存待执行的任务。backend 是结果存储（Redis/数据库），存任务执行结果。可以只配 broker 不配 backend（如果不需要查结果）。
如何保证 Celery 任务的幂等性？
任务可能因重试被多次执行。设计任务时让重复执行结果一致：用唯一约束防重复插入、用 update_or_create 代替 create、用分布式锁保护临界区。
自测三问
一个文章 API 要支持「按分类过滤、按标题搜索、按时间排序」，DRF 需要写哪些配置？
为什么 Celery 任务函数应该传 ID 而不是 ORM 对象？
ViewSet 的 get_queryset() 和 permission_classes 在权限控制上分别负责什么？
小结一句话 DRF 让 Django 变成专业的 API 框架（Serializer + ViewSet + Router），Celery 让耗时任务异步执行（Redis + Worker）；两者结合，Django 就能撑起现代前后端分离应用。
延伸