在响应后执行轻量任务

创建文章后还要写日志、发送通知。若这些动作都在返回响应前顺序完成，客户端要等得更久。本专题把轻量的附加工作挂到响应之后执行，并说明这种做法的边界。

BackgroundTask 表示一个任务，BackgroundTasks 可按顺序登记多个任务。它们仍在应用进程中运行，不能替代需要持久化和重试的任务队列。示例见下文。

同步附加工作的等待时间

先让处理函数同步完成日志操作，记录响应返回前发生了哪些事。

先看一个最小反例：

async def create_article(request):
    request_data = await request.json()
    title = request_data.get("title", "")

    article = {
        "id": 1,
        "title": title,
    }

    write_article_log(f"创建文章：{title}")

    return JSONResponse({
        "message": "文章创建成功",
        "article": article,
    })

问题不在语法，而在顺序：

先创建文章
再执行 write_article_log()
最后返回响应

如果日志、通知、清理都写在这里，用户必须等待它们全部完成。

用 BackgroundTask 写日志

给响应附上 BackgroundTask 后，客户端先获得响应，随后执行附加动作。运行脚本时还要检查日志结果。

目标：

创建文章
响应先返回
后台再记录日志

02_background_task.py：

from starlette.applications import Starlette
from starlette.background import BackgroundTask
from starlette.responses import JSONResponse
from starlette.routing import Route

def write_article_log(message):
    print(f"后台任务：{message}")

async def create_article(request):
    request_data = await request.json()
    title = request_data.get("title", "")

    article = {
        "id": 1,
        "title": title,
    }

    task = BackgroundTask(write_article_log, f"创建文章：{title}")

    return JSONResponse(
        {
            "message": "文章创建成功",
            "article": article,
        },
        background=task,
    )

routes = [
    Route("/api/articles", create_article, methods=["POST"]),
]

app = Starlette(routes=routes)

运行：

uvicorn 02_background_task:app --reload

可以用 requests 演示脚本向下面地址发送 POST 请求：

http://127.0.0.1:8000/api/articles

用 requests 发送 JSON 请求查看结果：

import requests

response = requests.post(
    "http://127.0.0.1:8000/api/articles",
    json={"title": "使用 BackgroundTask"},
    timeout=5,
)
print(response.status_code, response.json())

服务端终端随后会显示后台任务输出。

终端随后会输出：

后台任务：创建文章：使用 BackgroundTask

关键步骤只有三步：

1. 导入 BackgroundTask。
2. 创建任务：BackgroundTask(函数名, 参数...)。
3. 在响应对象上写 background=task。

注意不要写成：

task = BackgroundTask(write_article_log(f"创建文章：{title}"))

这会提前执行函数，不再是后台任务。

1. BackgroundTask 表示一个后台任务。
2. 任务函数可以接收参数。
3. 任务必须挂到响应对象上。
4. 响应发送后任务才会执行。

同步与异步任务

后台任务函数既可以是普通函数，也可以是异步函数。

例如：

async def send_article_notification(article_id, title):
    await asyncio.sleep(1)
    print(f"后台任务：发送文章通知，ID={article_id}，标题={title}")

但要记住：能异步，不代表适合做重任务。 Starlette 内置后台任务依然只适合轻量工作。

用 BackgroundTasks 安排多个任务

多个任务用 BackgroundTasks 登记。它们依次运行，前一个任务失败可能阻止后续任务执行。

现在希望创建文章后同时做两件事：

记录日志
发送通知

这时使用 BackgroundTasks。这一阶段的代码见 03_multiple_tasks.py；在示例目录运行 uvicorn 03_multiple_tasks:app --reload 后发送 POST 请求。

import asyncio

from starlette.applications import Starlette
from starlette.background import BackgroundTasks
from starlette.responses import JSONResponse
from starlette.routing import Route

ARTICLES = []

def write_article_log(message):
    print(f"后台任务：{message}")

async def send_article_notification(article_id, title):
    await asyncio.sleep(1)
    print(f"后台任务：发送文章通知，ID={article_id}，标题={title}")

async def create_article(request):
    request_data = await request.json()
    title = request_data.get("title", "")

    article = {
        "id": len(ARTICLES) + 1,
        "title": title,
    }
    ARTICLES.append(article)

    tasks = BackgroundTasks()
    tasks.add_task(write_article_log, f"创建文章：{title}")
    tasks.add_task(send_article_notification, article["id"], article["title"])

    return JSONResponse(
        {
            "message": "文章创建成功",
            "article": article,
        },
        background=tasks,
    )

运行后，可以向下面地址发送 POST 请求观察效果：

http://127.0.0.1:8000/api/articles

关键点：

先创建 BackgroundTasks()
用 add_task() 逐个加入任务
最后通过 background=tasks 挂到响应对象

BackgroundTasks 的适用时机

判断很简单：

一个任务 → BackgroundTask
多个任务 → BackgroundTasks

后台任务的适用边界

适合	不适合
简单日志	很耗时的数据处理
简单通知	必须可靠送达的消息
清理临时文件	需要失败重试的任务
简单埋点	分布式任务
响应后的小动作	长时间批处理

适合的场景

共同特点：

不影响本次响应能否成立
用户不必等待
任务本身很轻

不适合的场景

如果任务需要下面这些能力，就不要只依赖内置后台任务：

可靠执行
失败重试
进度查询
长时间运行
分布式调度

后台任务与其他机制的分工

机制	执行时机	职责
Lifespan	应用启动和关闭时	初始化和清理资源
中间件	每个请求前后	统一处理请求和响应
BackgroundTask	某个响应之后	一个轻量后续动作
BackgroundTasks	某个响应之后	多个轻量后续动作
Endpoint	请求处理中	当前接口的主要业务

可以这样记：

Lifespan 管应用生命周期
中间件管请求公共流程
Endpoint 管当前业务
后台任务管响应后的轻量补充动作

文章创建后的日志与通知

完整案例只保留本专题需要的文章 API。发送 POST /api/articles 后，处理函数立即创建文章并构造 201 响应；两个后台任务按登记顺序写日志和打印通知。已有文章也可以单独安排通知。POST /api/cleanup-temp 展示响应后清理临时文件的用法。

tasks = BackgroundTasks()
tasks.add_task(write_article_log, f"创建文章：{title}")
tasks.add_task(send_notification, article)
return JSONResponse({"article": article}, status_code=201, background=tasks)

在 _code_starlette/p03c02 中启动 uvicorn 04_article_tasks:app --reload，另开终端运行 python background_tasks_demo.py。演示脚本用 requests 依次创建文章、安排通知和清理临时文件，并打印状态码与响应体。服务端终端会显示后台任务输出；logs/article.log 中会新增创建记录。

观察响应时机要区分客户端类型：经网络访问时，响应体发送后才执行这些任务；TestClient 为了完成一次请求，会等待后台任务结束。因此不要用 TestClient 的耗时来推断真实网络客户端拿到响应的时间。

这些任务仍在应用进程内。进程中断、任务失败或需要可靠重试时，应改用独立任务系统；本案例只演示轻量的响应后工作。

案例源码

01_blocking_task.py

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route


def write_article_log(message):
    print(f"写入日志：{message}")


async def create_article(request):
    request_data = await request.json()
    title = request_data.get("title", "")

    article = {
        "id": 1,
        "title": title,
    }

    write_article_log(f"创建文章：{title}")

    return JSONResponse(
        {
            "message": "文章创建成功",
            "article": article,
        }
    )


routes = [
    Route("/api/articles", create_article, methods=["POST"]),
]


app = Starlette(routes=routes)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("01_blocking_task:app", host="127.0.0.1", port=8000, reload=True)

02_background_task.py

import asyncio

from starlette.applications import Starlette
from starlette.background import BackgroundTask
from starlette.responses import JSONResponse
from starlette.routing import Route


ARTICLES = [
    {"id": 1, "title": "认识 Starlette 应用对象"},
    {"id": 2, "title": "学习后台任务"},
]


def find_article(article_id):
    for article in ARTICLES:
        if article["id"] == article_id:
            return article
    return None


def write_article_log(message):
    print(f"后台任务：{message}")


async def send_article_notification(article_id, title):
    await asyncio.sleep(1)
    print(f"后台任务：发送文章通知，ID={article_id}，标题={title}")


async def create_article(request):
    request_data = await request.json()
    title = request_data.get("title", "")

    article = {
        "id": 1,
        "title": title,
    }

    task = BackgroundTask(write_article_log, f"创建文章：{title}")

    return JSONResponse(
        {
            "message": "文章创建成功",
            "article": article,
        },
        background=task,
    )


async def notify_article(request):
    article_id = request.path_params["article_id"]
    article = find_article(article_id)

    if article is None:
        return JSONResponse(
            {"message": "文章不存在"},
            status_code=404,
        )

    task = BackgroundTask(
        send_article_notification,
        article["id"],
        article["title"],
    )

    return JSONResponse(
        {
            "message": "通知任务已提交",
            "article": article,
        },
        background=task,
    )


routes = [
    Route("/api/articles", create_article, methods=["POST"]),
    Route("/api/articles/{article_id:int}/notify", notify_article, methods=["POST"]),
]


app = Starlette(routes=routes)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("02_background_task:app", host="127.0.0.1", port=8000, reload=True)

03_multiple_tasks.py

import asyncio

from starlette.applications import Starlette
from starlette.background import BackgroundTasks
from starlette.responses import JSONResponse
from starlette.routing import Route


ARTICLES = []


def write_article_log(message):
    print(f"后台任务：{message}")


async def send_article_notification(article_id, title):
    await asyncio.sleep(1)
    print(f"后台任务：发送文章通知，ID={article_id}，标题={title}")


async def create_article(request):
    request_data = await request.json()
    title = request_data.get("title", "")

    article = {
        "id": len(ARTICLES) + 1,
        "title": title,
    }
    ARTICLES.append(article)

    tasks = BackgroundTasks()
    tasks.add_task(write_article_log, f"创建文章：{title}")
    tasks.add_task(send_article_notification, article["id"], article["title"])

    return JSONResponse(
        {
            "message": "文章创建成功",
            "article": article,
        },
        background=tasks,
    )


routes = [
    Route("/api/articles", create_article, methods=["POST"]),
]


app = Starlette(routes=routes)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("03_multiple_tasks:app", host="127.0.0.1", port=8000, reload=True)

04_article_tasks.py

import asyncio
from pathlib import Path

from starlette.applications import Starlette
from starlette.background import BackgroundTask, BackgroundTasks
from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse
from starlette.routing import Route


ARTICLES = []
BASE_DIR = Path(__file__).parent
LOG_FILE = BASE_DIR / "logs" / "article.log"
TEMP_FILE = BASE_DIR / "temp" / "demo.txt"


def write_article_log(message):
    LOG_FILE.parent.mkdir(exist_ok=True)
    with LOG_FILE.open("a", encoding="utf-8") as log:
        log.write(message + "\n")
    print(f"后台写入日志：{message}")


async def send_notification(article):
    await asyncio.sleep(0.2)
    print(f"后台发送通知：文章 {article['id']}《{article['title']}》")


def cleanup_temp_file():
    TEMP_FILE.unlink(missing_ok=True)
    print("后台清理临时文件")


async def list_articles(request):
    return JSONResponse({"items": ARTICLES})


async def create_article(request):
    data = await request.json()
    title = str(data.get("title", "")).strip()
    if not title:
        raise HTTPException(status_code=400, detail="文章标题不能为空")

    article = {"id": len(ARTICLES) + 1, "title": title}
    ARTICLES.append(article)

    tasks = BackgroundTasks()
    tasks.add_task(write_article_log, f"创建文章：{title}")
    tasks.add_task(send_notification, article)
    return JSONResponse({"article": article}, status_code=201, background=tasks)


async def notify_article(request):
    article_id = request.path_params["article_id"]
    article = next((item for item in ARTICLES if item["id"] == article_id), None)
    if article is None:
        raise HTTPException(status_code=404, detail="文章不存在")
    task = BackgroundTask(send_notification, article)
    return JSONResponse({"message": "通知任务已安排"}, background=task)


async def cleanup_demo(request):
    TEMP_FILE.parent.mkdir(exist_ok=True)
    TEMP_FILE.write_text("等待清理", encoding="utf-8")
    task = BackgroundTask(cleanup_temp_file)
    return JSONResponse({"message": "清理任务已安排"}, background=task)


app = Starlette(routes=[
    Route("/api/articles", list_articles, methods=["GET"]),
    Route("/api/articles", create_article, methods=["POST"]),
    Route("/api/articles/{article_id:int}/notify", notify_article, methods=["POST"]),
    Route("/api/cleanup-temp", cleanup_demo, methods=["POST"]),
])

background_tasks_demo.py

import requests


BASE_URL = "http://127.0.0.1:8000"


def test_create_article():
    response = requests.post(
        f"{BASE_URL}/api/articles",
        json={"title": "后台任务实战"},
        timeout=5,
    )
    print("POST /api/articles")
    print("状态码：", response.status_code)
    print("响应体：", response.json())


def test_notify_article():
    response = requests.post(f"{BASE_URL}/api/articles/1/notify", timeout=5)
    print("POST /api/articles/1/notify")
    print("状态码：", response.status_code)
    print("响应体：", response.json())


def test_cleanup_temp():
    response = requests.post(f"{BASE_URL}/api/cleanup-temp", timeout=5)
    print("POST /api/cleanup-temp")
    print("状态码：", response.status_code)
    print("响应体：", response.json())


if __name__ == "__main__":
    test_create_article()
    test_notify_article()
    test_cleanup_temp()