后端Web框架【免费下载链接】starletteThe little ASGI framework that shines. 项目地址https://gitcode.com/gh_mirrors/st/starlette点击查看免费下载Starlette 内置的BackgroundTask/BackgroundTasks是在进程内执行响应之后才需要完成的工作的官方方案先快速把 HTTP 响应返回给客户端再在后台执行发邮件、写日志、通知管理员等耗时操作。本文以官方文档 docs/background.md 为主体结合 starlette/background.py 的实现细节与 tests/test_background.py 的测试用例讲清后台任务的创建、附加、执行时机、同步/异步任务的处理方式以及异常传播语义帮助你在真实应用中安全、正确地使用这一特性。一、什么是进程内后台任务Web 应用中常有一些用户不关心结果、但不得不做的操作注册成功后发送欢迎邮件、同步数据到第三方服务、记录审计日志。如果把这些操作放在请求处理流程中同步执行接口的响应时间会被无谓拖长。Starlette 的答案是进程内后台任务in-process background task由BackgroundTask类提供定义在 starlette/background.py必须附加到某个 Response 对象上通过响应的background参数传入只在响应已经发送之后才执行因此客户端无需等待任务完成即可收到响应。需要特别注意的是这类任务是进程内的它只在当前运行的应用进程里执行一旦进程退出任务即丢失。它适合短时、非关键路径的收尾工作对于需要持久化、跨进程调度的场景应使用任务队列等外部方案。二、BackgroundTask为响应挂载单个后台任务2.1 签名与基本用法BackgroundTask的构造函数签名见 starlette/background.py为BackgroundTask(func, *args, **kwargs)func要执行的可调用对象可以是async 函数也可以是普通同步函数Starlette 会自动在线程池中执行它*args、**kwargs传给func的位置参数和关键字参数。官方文档给出了最典型的注册场景用户注册成功后在响应发出后再发送欢迎邮件。以下是文档中 docs/background.md 示例的完整可运行版本from starlette.applications import Starlette from starlette.responses import JSONResponse from starlette.routing import Route from starlette.background import BackgroundTask async def signup(request): data await request.json() username data[username] email data[email] task BackgroundTask(send_welcome_email, to_addressemail) message {status: Signup successful} return JSONResponse(message, backgroundtask) async def send_welcome_email(to_address): ... routes [ Route(/user/signup, endpointsignup, methods[POST]) ] app Starlette(routesroutes)关键点是第 14~15 行后台任务在创建响应时通过backgroundtask参数与响应绑定任务里的send_welcome_email会收到to_addressemail这个关键字参数。2.2 同步函数也可以直接作为任务BackgroundTask对同步函数同样友好。构造时它会调用is_async_callable(func)判断func是否为协程函数见 starlette/background.py运行时再分派async def __call__(self) - None: if self.is_async: await self.func(*self.args, **self.kwargs) else: await run_in_threadpool(self.func, *self.args, **self.kwargs)async 任务直接await同步任务通过run_in_threadpool在线程池中执行不会阻塞事件循环。run_in_threadpool的底层实现位于 starlette/concurrency.py本质是对anyio.to_thread.run_sync的封装async def run_in_threadpool(func: Callable[P, T], *args: P.args, **kwargs: P.kwargs) - T: func functools.partial(func, *args, **kwargs) return await anyio.to_thread.run_sync(func)也就是说你在后台任务里写普通同步代码比如调用requests、smtplib是完全可行的——这些阻塞操作会被放到线程池中不会卡住 ASGI 事件循环。这一行为在测试中也有覆盖tests/test_background.py里的test_sync_task验证了同步任务同样会在响应发送后执行。2.3 参数在构造时捕获BackgroundTask在构造时就把args/kwargs保存下来starlette/background.py执行时统一传给func。因此你可以放心在循环或闭包中逐个构造任务每个任务实例都持有自己独立的一份参数快照不会出现闭包变量共享的经典坑。三、BackgroundTasks按顺序执行多个后台任务当需要在一个响应上挂载多个后台任务时使用BackgroundTasks其签名见 starlette/background.py为BackgroundTasks(tasks[])tasks可选初始化时可传入一个BackgroundTask序列更常用的方式是通过add_task(func, *args, **kwargs)逐个添加任务内部会自动包装成BackgroundTask实例class BackgroundTasks(BackgroundTask): def __init__(self, tasks: Sequence[BackgroundTask] | None None): self.tasks list(tasks) if tasks else [] def add_task(self, func: Callable[P, Any], *args: P.args, **kwargs: P.kwargs) - None: task BackgroundTask(func, *args, **kwargs) self.tasks.append(task)注意BackgroundTasks继承自BackgroundTask因此它本身也可以像单个任务一样被响应接受和调用。官方文档 docs/background.md 中的多任务示例注册成功后既给用户发欢迎邮件又给管理员发通知from starlette.applications import Starlette from starlette.responses import JSONResponse from starlette.background import BackgroundTasks async def signup(request): data await request.json() username data[username] email data[email] tasks BackgroundTasks() tasks.add_task(send_welcome_email, to_addressemail) tasks.add_task(send_admin_notification, usernameusername) message {status: Signup successful} return JSONResponse(message, backgroundtasks) async def send_welcome_email(to_address): ... async def send_admin_notification(username): ... routes [ Route(/user/signup, endpointsignup, methods[POST]) ] app Starlette(routesroutes)3.1 严格按添加顺序执行BackgroundTasks的执行逻辑非常直接starlette/background.pyasync def __call__(self) - None: for task in self.tasks: await task()任务按添加顺序依次串行执行而不是并发执行。文档特别强调了一条重要语义任务按顺序执行一旦某个任务抛出异常后续任务将不再有机会执行。这一行为被 tests/test_background.py 中的test_multi_tasks_failure_avoids_next_execution显式验证两个任务都执行increment但第一个任务在执行时抛出Exception(task failed)最终断言TASK_COUNTER 1即第二个任务确实没有被执行且异常会从响应调用中向外抛出。因此在设计多任务时请务必将有依赖关系的任务放在后面、独立且重要的任务放在前面并在任务函数内部自行捕获可容忍的异常避免一个失败的任务连坐后续所有任务。四、响应发送后执行的完整机制4.1 哪些响应支持 background 参数background参数是Response基类构造函数的一部分starlette/responses.py因此所有响应类型都支持JSONResponse、HTMLResponse、PlainTextResponse、RedirectResponse、StreamingResponse、FileResponse等。从源码可见RedirectResponse也接受background参数starlette/responses.py这意味着你可以在重定向用户的同时执行后台任务。4.2 执行时机响应已发送之后普通Response的执行流程在 starlette/responses.pyasync def __call__(self, scope, receive, send) - None: if scope[type] websocket: send self._wrap_websocket_denial_send(send) await send({type: http.response.start, status: self.status_code, headers: self.raw_headers}) await send({type: http.response.body, body: self.body}) if self.background is not None: await self.background()先通过send将响应头和响应体发回客户端随后才调用await self.background()。这正是任务只在响应已经发送后运行承诺的落地实现。对于流式响应执行时机略有不同但原则一致StreamingResponse在流式发送完整个 body包括处理完more_body的所有数据块之后才执行后台任务FileResponse也是在文件内容发送完成后触发starlette/responses.py。4.3 与中间件的交互BaseHTTPMiddleware内部也有对应的后台任务支持其_StreamingResponse在流结束时同样会执行绑定的后台任务见 starlette/middleware/base.py。这一区域历史上经过多轮修复docs/release-notes.md 中有明确记录修复了使用BaseHTTPMiddleware且客户端断开时BackgroundTasks被取消的问题PR #1715支持在BaseHTTPMiddleware内部调度BackgroundTasksPR #2688修复BaseHTTPMiddleware场景下后台任务异常的上抛PR #2812。如果你在中间件中创建响应并挂载后台任务当前版本的行为是有测试与修复保障的。五、后台任务中的异常处理语义后台任务的异常处理有一个容易踩坑的特殊语义官方文档在 docs/exceptions.md 中专门说明如果BackgroundTask抛出异常它会被handle_error函数处理但此时响应已经发送因此handle_error返回的响应对象会被丢弃。如果错误发生在响应发送之前则会正常使用错误处理器返回的响应。也就是说响应已经送达客户端错误处理器不能再重发一个错误响应——客户端拿到的始终是正常的成功响应后台任务的异常不会静默吞掉它会沿调用链抛出最终交给 Starlette 的错误处理机制ServerErrorMiddleware等记录或处理对于多个任务正如上文所述异常还会中断后续任务的执行。所以务实的做法是在任务函数内部用try/except包裹可能失败的子操作把任务级异常消化在任务内部避免影响其他任务和整体的错误日志噪音。六、后台任务与应用生命周期的关系后台任务不是无限期的应用关闭时Starlette 会等它们完成。官方文档 docs/lifespan.md 明确指出lifespan 的 teardown 会在所有连接关闭、且所有进程内后台任务完成后运行。这保证了优雅关闭语义即使收到停机信号已经排队的进程内后台任务也有机会跑完然后才执行 lifespan 中yield之后的关闭逻辑。如果你需要在后台任务之外管理更复杂的并发任务文档建议使用anyio.create_task_group()这类任务组原语。七、适用场景与使用建议基于以上原理总结出几条实践建议适合发送通知邮件、调用不重要的第三方回调、写审计日志、清理临时资源等响应后收尾工作。不适合任务超长、需要持久化、需要跨进程/跨机器执行的场景——进程内任务随进程消亡而丢失此时应引入外部任务队列。同步任务放心写同步函数会自动跑在线程池中不会阻塞事件循环但要注意线程池资源同样有限避免堆叠大量长时间同步任务。多个任务注意顺序与容错任务按添加顺序串行执行前一个抛异常则后续全部不执行把不重要的任务放前面或在任务内部自行捕获异常。异常不要期待能改响应后台任务运行时响应已发送错误处理器返回的响应会被丢弃应通过日志或指标观测后台任务失败。后台任务是一个精巧的响应优先、收尾后置机制它让你的接口快速返回同时不放弃那些必须完成的收尾工作。结合 tests/test_background.py 中的测试用例异步任务、同步任务、多任务顺序、异常中断你可以为上述每一条语义找到可验证的行为依据在项目中放心使用。赞分享后端Web框架【免费下载链接】starletteThe little ASGI framework that shines. 项目地址https://gitcode.com/gh_mirrors/st/starlette点击查看免费下载相关推荐Litestar 后台任务BackgroundTask / BackgroundTasks完整指南响应发送后的异步收尾Litestar 后台任务BackgroundTask / BackgroundTasks完整指南响应发送后的异步收尾 Litestar 内置的后台任务机后端Web框架FastAPI 后台任务BackgroundTasks实战指南响应后执行异步任务的完整方案FastAPI 后台任务BackgroundTasks实战指南响应后执行异步任务的完整方案 本文围绕 FastAPI 官方文档 docs/pt/docs/后端Web框架API设计Sanic 后台任务完全指南add_task 的用法、原理与最佳实践Sanic 后台任务完全指南add_task 的用法、原理与最佳实践 导读 在异步 Web 框架中把一段协程丢到事件循环里跑是实现定时通知、消息推送、后端Web框架创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
