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

异步视图与 Tasks 框架

吴秀峰 2026-11-01 3 次阅读 0 条评论 ↓

适用 Django: 6.1.2
适用 Python: 3.14（兼容 3.12–3.14）
学习难度: 入门
最近核验: 2026-11-01

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

学习目标

区分异步 I/O、后台任务和多进程服务器，运行 async ORM 及 Django Tasks 示例，清楚说明当前例子的生产边界。

为什么 async 不等于“所有代码更快”

异步适合等待网络等 I/O 时让出执行机会，不能让纯 CPU 计算自动加速。完整异步请求链需要 ASGI 服务器和兼容中间件；WSGI 下运行 async 视图仍有适配开销。数据库事务操作目前应封装在同步函数里，再通过正确的桥接方式调用，不能把 synchronous_only 错误用关闭保护的方式压过去。

本章异步视图使用 acount 查询公开文章数量，只演示接口方式。本站和本教程部署主线仍采用同步 Gunicorn WSGI，适合当前博客负载；没有为了展示语法更换生产服务器。

Django 6.0 起的 Tasks 提供任务定义和提交接口，不自带生产 worker。ImmediateBackend 在当前进程立即执行，适合开发和测试；这不降低请求耗时，也不是持久化队列。生产要选择兼容后端和独立 worker，并设计重试、超时、幂等和失败监控。

具体怎么做

新建 async_views.py，注册只读计数接口。
新建 tasks.py，使用 @task 定义一个无副作用的计数任务。
配置开发用 ImmediateBackend，在 shell 中执行 enqueue 并读取结果。
不把这个任务接入写文章的生产请求路径，避免把“立即执行”误解为后台加速。

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

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

`blog/async_views.py`

from django.http import JsonResponse
from django.views.decorators.http import require_GET
from .models import Post

@require_GET
async def public_count(request):
    count = await Post.objects.filter(status=Post.Status.PUBLISHED).acount()
    return JsonResponse({"published": count})

`blog/tasks.py`

from django.tasks import task
from .models import Post

@task
def count_public_posts():
    # 只读、可重复执行。不要把密码、令牌或 request 对象作为任务参数。
    return Post.objects.filter(status=Post.Status.PUBLISHED).count()

`blog/urls.py`

from django.urls import path
from . import views, api, async_views
app_name = "blog"
urlpatterns = [
    path("api/public-count/", async_views.public_count, name="async_count"),
    path("api/posts/", api.posts, name="api_posts"),
    path("api/posts/<int:pk>/", api.detail, name="api_detail"),
    path("", views.home, name="home"),
    path("mine/", views.mine, name="mine"),
    path("posts/new/", views.create, name="create"),
    path("posts/<int:pk>/comments/", views.comment, name="comment"),
    path("posts/<int:pk>/edit/", views.edit, name="edit"),
    path("posts/<int:pk>/delete/", views.delete, name="delete"),
    path("posts/<int:pk>/", views.detail, name="detail"),
    path("about/", views.about, name="about"),
    path("old-home/", views.old_home, name="old_home"),
]

`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.middleware.csp.ContentSecurityPolicyMiddleware",
    "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.template.context_processors.csp",
        "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"}}

LOGGING = {
    "version": 1, "disable_existing_loggers": False,
    "formatters": {"standard": {"format": "{asctime} {levelname} {name} {message}", "style": "{"}},
    "handlers": {"console": {"class": "logging.StreamHandler", "formatter": "standard"}},
    "root": {"handlers": ["console"], "level": "INFO"},
}

from django.utils.csp import CSP
SECURE_CSP = {
    "default-src": [CSP.SELF], "script-src": [CSP.SELF, CSP.NONCE],
    "style-src": [CSP.SELF, CSP.NONCE], "img-src": [CSP.SELF, "data:"],
    "object-src": [CSP.NONE], "base-uri": [CSP.SELF], "frame-ancestors": [CSP.NONE],
    "form-action": [CSP.SELF],
}
SESSION_COOKIE_HTTPONLY = True
SECURE_CONTENT_TYPE_NOSNIFF = True
X_FRAME_OPTIONS = "DENY"

# 仅用于本章演示：ImmediateBackend 在调用进程立即执行，不是生产队列。
TASKS = {"default": {"BACKEND": "django.tasks.backends.immediate.ImmediateBackend"}}

`examples/run_task.py`

from blog.tasks import count_public_posts
result = count_public_posts.enqueue()
print("任务状态：", result.status)
print("公开文章数量：", result.return_value)

如何确认成功

python manage.py shell -c "exec(open('examples/run_task.py').read())"
curl http://127.0.0.1:8000/api/public-count/

任务状态应成功，并输出公开文章数量；异步接口返回同样口径的 published。后台日志可看到任务开始与结束，执行仍发生在本次 shell 进程。

练习：设计“发布后发通知”的任务参数，只传文章 id 和事件 id，不传 request、模型对象或密码。说明为什么应在事务提交后入队，为什么重复执行不能重复给用户发送大量邮件，以及队列故障时如何补偿。

常见问题

换成 async def 就解决慢查询了吗？ 没有，索引和查询规模仍重要。DummyBackend 会执行任务吗？ 不会，它主要服务测试，不能当作真实队列。Celery 自动支持 Django Tasks 接口吗？ 不能这样假定，需要具体的适配后端与版本验证。生产一定需要任务队列吗？ 小博客可以先保持同步和明确超时，实际出现长耗时任务再引入基础设施。

官方参考

把学习连起来

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

登录并保存学习进度 →

联系站长

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

学习目标

为什么 async 不等于“所有代码更快”

具体怎么做

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

blog/async_views.py

blog/tasks.py

blog/urls.py

config/settings.py

examples/run_task.py