在Python的Web框架生态里Starlette是个很有意思的存在。它本身不追求大而全却撑起了很多上层框架的根基。做异步Web服务时我经常直接拿它当底子来搭省心又可控。如果你已经用过FastAPI那其实你早就在间接使用Starlette的那套路由和请求处理逻辑如果你还没接触过那这篇文章正好可以带你以最快的路径搞懂这层地基到底是怎么工作的。Starlette解决的痛点很直接Web框架要么太重自带一堆你用不上的东西要么太老对ASGI这类异步协议支持得磕磕绊绊。而Starlette只保留最核心的Web能力——路由、中间件、WebSocket、后台任务、测试支持其余的统统不管需要什么自己装什么。所以它特别适合三类人想从底层理解异步Web框架如何工作的学习者、需要一个精简可控的服务骨架来承载内部API的开发者、以及想给FastAPI这类上层框架做定制扩展的进阶用户。下面没有废话直接带你过一遍从原理到落地的完整链路。1. 先搞清楚ASGI再看Starlette的位置1.1 从WSGI到ASGI异步Web接口为什么会出现在Python Web开发的历史里最早被广泛接受的通用接口是WSGI。Django、Flask这些老牌框架本质上是WSGI应用它们接收一个environ字典和start_response回调返回一个可迭代的响应体。这套设计的核心假设是同步模型一个worker在一个时间点只处理一个请求想并发就开多线程或多进程。短请求场景下WSGI完全够用但WebSocket、Server-Sent Events这类长连接需求一出现它就力不从心了。WebSocket连接建立后要一直保持服务端随时能主动推送消息SSE要求服务端持续输出事件流HTTP/2协议还引入了多路复用的概念。这些场景里请求-响应-结束的整体流程根本套不住因为连接本身的价值就在于它一直活着。ASGIAsynchronous Server Gateway Interface就是为此出现的。ASGI应用是一个异步可调用对象接收三个参数scope、receive、send。scope里装着连接的基本信息receive用于异步接收事件send用于异步发送事件。应用内部可以自己维护状态、自己决定什么时候收发消息连接的生命周期完全由应用掌控。我把这个差异比作传统电话总机和现代客服系统。WSGI像老式接线台每条线路要分配给一个接线员通话结束才释放ASGI像一个客服平台同一批客服可以同时服务大量会话每个会话状态互相独立靠事件驱动来回切换。这个区别直接决定了Starlette这类框架的设计方向它能同时维护大量长连接CPU占用小并且在单进程内就能做异步并发非常适合IO密集型服务。理解了这一点你就知道Starlette为什么敢把轻量作为核心卖点。1.2 Starlette的设计哲学只做Web层该做的事Starlette的定位可以总结为三句话轻、模块化、可替换。安装它不会带任何强制依赖核心就是路由、请求响应对象、中间件、WebSocket这些ASGI应用的基本构成。它不像很多框架那样捆绑ORM、模板引擎、表单校验、管理后台这些统统留给使用者自己决定。这种设计的直接好处是你可以只用它的某一个组件。比如只拿Router做路由分发或者只拿TestClient做接口测试不必启动一整个Web框架。这种模块化在工程上很实用我甚至在一些工具脚本里单独用过它的Response类只为了统一格式返回数据。从FastAPI入手的人可能感受最深。FastAPI在Starlette之上增加了pydantic数据校验和OpenAPI文档生成而路由、请求、响应这些底层能力其实都来自Starlette。所以学Starlette并不是学一套和FastAPI无关的东西而是在把上层框架的引擎盖打开看一眼内部构造。我做了个简易对比表方便你快速定位它和常见框架的关系维度StarletteFastAPIFlask协议ASGI异步ASGI异步WSGI同步数据校验无可外部集成内置pydantic无可外部集成自动文档无有OpenAPI无WebSocket内置支持内置支持需扩展后台任务内置内置需扩展依赖注入无内置无最小依赖量几乎零有额外依赖少量依赖从这个表能看出来Starlette其实比FastAPI更底层比Flask更适合做异步服务。它不是用来取代谁的而是给你一个更贴近协议层的选择空间。2. 框架选型哪些场景值得直接上Starlette2.1 当FastAPI太重、Flask太老时选框架本质是在选约束。FastAPI非常适合需要对外提供规范API、看重自动文档和类型校验的场景。但如果你并不需要导出OpenAPI文档或者团队已经有了一整套自己的序列化校验方案那FastAPI这层皮就是多余的。反观Starlette它不会替你决定数据校验用什么、日志怎么打、配置怎么管理你需要什么就自己接什么。这种控制力在两类项目里特别有价值一类是内部服务接口文档是给同事看的不需要自动生成另一类是网关或代理层需要自己控制请求生命周期、做自定义路由和转发逻辑。Flask在轻量框架里有着极高的占有率但它停在WSGI时代。在Flask里想处理WebSocket你得额外装扩展而且这些扩展大多不是官方维护质量和兼容性参差不齐。而Starlette原生就把WebSocket纳入核心能力路由、接收、发送都是标准接口不用绕弯子。我之前帮一个团队做消息推送服务他们最初用的Flask加第三方WebSocket插件到了高并发阶段出现大量连接不稳定问题。后来我们只把接入层换成了Starlette连接管理和消息广播逻辑反而变简单了问题也少了。这就是异步协议支持从底层就做好的价值。2.2 几个典型的落地场景根据我平时观察和踩过的坑下面这些场景直接使用Starlette会比较得心应手内部API服务只有几个端点给前端或内部系统调用不需要复杂框架。异步爬虫或数据管道后台服务请求入口和业务逻辑分离Starlette只负责HTTP入口业务在其它协程任务里跑。WebSocket网关做消息推送、在线聊天、实时协作等场景原生WebSocket支持非常方便。组件独立使用单独拿Router管理路由、单独拿TestClient做接口测试、用FileResponse提供文件下载。注意区分用到Starlette和用Starlette做框架如果你用FastAPI实际上已经用到了Starlette的底层能力。所以我说它是地基这一点也不夸张。2.3 选型时的一个自检清单我一般会问自己四个问题基本能判断该不该手写Starlette需要自动生成OpenAPI文档吗需要就选FastAPI不需要可以考虑Starlette。数据校验、依赖注入这些能力有现成方案吗有并且不想换直接用Starlette做外部集成没有选FastAPI更省事。需要处理WebSocket或长连接吗需要的话Starlette的加分项非常明显。团队是不是已经很熟悉某个框架的扩展生态如果大家重度依赖Flask的某个插件迁移成本要想清楚。这个清单不绝对但能帮你避免一开始就选错方向。我自己通常的做法是完整对外服务用FastAPI内部小服务或网关层才手写Starlette。3. 手把手搭建核心服务路由、中间件、响应与任务3.1 安装与最小可运行应用安装非常简单只是注意Starlette本身不带服务器需要配一个ASGI serverpip install starlette uvicornuvicorn是常用的ASGI服务器启动命令长这样uvicorn app:app --host 0.0.0.0 --port 8000最小应用代码from starlette.applications import Starlette from starlette.responses import JSONResponse from starlette.routing import Route async def home(request): return JSONResponse({message: hello starlette}) app Starlette(routes[ Route(/, home), ])这个app本身就是一个可调用对象实现了async def __call__(scope, receive, send)。uvicorn把ASGI三要素传进来Starlette内部负责路由匹配、中间件执行、请求对象构建最后把响应发回去。初看可能觉得绕但这就是ASGI应用的标准形态比WSGI的environ加start_response更直观。3.2 路由、路径参数与请求处理路由是Web框架的基本盘。Starlette的Route接收路径和端点函数路径里支持参数和类型转换async def user_detail(request): user_id request.path_params[user_id] return JSONResponse({user_id: user_id}) async def product_detail(request): product_id request.path_params[product_id] return JSONResponse({product_id: product_id}) app Starlette(routes[ Route(/users/{user_id}, user_detail), Route(/products/{product_id:int}, product_detail), ]){user_id}是字符串参数{product_id:int}会自动转成int如果传了非数字会直接返回404。类似地还支持{path:path}用于匹配包含斜杠的完整路径。同一个路径对应多个HTTP方法通过methods参数控制async def handle(request): if request.method GET: return JSONResponse({method: get}) if request.method POST: body await request.json() return JSONResponse({received: body}) Route(/items, handle, methods[GET, POST])请求对象里最常用的就是request.path_params、request.query_params、request.headers、await request.body()、await request.json()。查询参数是MultiDict类型取多个值的时候和普通dict行为不太一样这个细节在参数多时容易踩坑后面避坑章节会专门说。3.3 中间件的写法与执行顺序中间件在Starlette里是典型的洋葱模型请求从外层进入经过所有中间件到达端点后再从内层向外返回。先看一个自定义计时中间件from starlette.middleware.base import BaseHTTPMiddleware import time class TimingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): start time.perf_counter() response await call_next(request) duration time.perf_counter() - start response.headers[x-process-time] f{duration:.4f} return response app Starlette( routes[Route(/, home)], middleware[ Middleware(TimingMiddleware), Middleware(CORSMiddleware, allow_origins[*]), ], )BaseHTTPMiddleware是官方提供的简化封装call_next就是请求往下传的入口。新版文档里也鼓励写纯ASGI中间件因为BaseHTTPMiddleware在某些高并发场景下性能不占优但普通项目用起来差别不大。中间件顺序很关键。异常处理中间件要放在最外层这样它才能捕获到后面所有中间件和端点抛出的异常。日志、计时这类中间件放在里层减少对外层状态的影响。跨域中间件一般也放最外层否则非简单请求的预检可能都到不了业务代码。官方内置了几个实用中间件值得记住CORSMiddleware跨域访问控制GZipMiddleware对响应做gzip压缩SessionMiddleware提供签名Cookie会话TrustedHostMiddlewareHost头校验防DNS rebinding3.4 响应对象的选择Starlette提供了好几种Response选对能少写不少代码。日常最常见的JSONResponse直接返回字典或列表最常用。HTMLResponse渲染一个HTML字符串。PlainTextResponse返回纯文本。StreamingResponse流式返回适合大文件传输或流式生成内容。RedirectResponse重定向。FileResponse文件下载支持range请求和断点续传。流式返回是一个很容易被忽视的能力看这个例子async def streaming(request): async def generator(): for i in range(100): yield fdata: {i}\n await asyncio.sleep(0.1) return StreamingResponse(generator(), media_typetext/plain)generator必须是异步生成器StreamingResponse会按迭代逐个发送。加上asyncio.sleep模拟耗时操作客户端可以看到内容按块接收这就很适合SSE推送和大文件下载。3.5 后台任务响应发完再干活后台任务解决的是用户请求已经完成但还有事情要收尾的问题。常见场景包括发通知、写日志、清理临时资源。Starlette提供了BackgroundTask和BackgroundTasks两种方式from starlette.background import BackgroundTask, BackgroundTasks def notify(user_id: str): # 模拟消息推送或日志落库 with open(notify.log, a) as f: f.write(fnotify user {user_id}\n) async def create_user(request): payload await request.json() task BackgroundTask(notify, payload[id]) return JSONResponse({ok: True}, backgroundtask)这里有个很关键的语义差别BackgroundTask传入普通函数时它在线程池执行传入async函数时它在事件循环执行。任务状态和请求上下文完全隔离所以别指望在任务里拿到当前Request对象需要的数据要在创建任务时就传进去。3.6 WebSocket拉开差距的核心能力WebSocket是Starlette对比传统WSGI框架最明显的优势。核心API不多先看最基础的echo实现from starlette.websockets import WebSocket, WebSocketDisconnect async def chat(websocket: WebSocket): await websocket.accept() try: while True: message await websocket.receive_text() await websocket.send_text(fecho: {message}) except WebSocketDisconnect: print(client disconnected)accept()建立连接receive_text()阻塞等待客户端消息send_text()发送消息客户端断开时会抛出WebSocketDisconnect异常。除此之外还有receive_bytes/send_bytes、receive_json/send_json按你的消息格式选择。这里最重要的是长连接状态管理。每个WebSocket连接在应用层是一个独立对象但通常你会同时服务多个客户端。所以我在项目里一般用一个连接管理器类统一维护连接集合后面实操章节会完整展开。4. 完整实操用Starlette搭一个多房间在线聊天服务4.1 项目结构和连接管理器这节我们做一个支持多房间、多用户同时在线的聊天服务。业务分两块REST接口负责房间管理WebSocket负责消息收发。先定义连接管理器from typing import Dict, List from starlette.websockets import WebSocket, WebSocketState, WebSocketDisconnect class RoomManager: def __init__(self): self.rooms: Dict[str, List[WebSocket]] {} async def connect(self, room_id: str, ws: WebSocket): await ws.accept() self.rooms.setdefault(room_id, []).append(ws) def disconnect(self, room_id: str, ws: WebSocket): conns self.rooms.get(room_id, []) if ws in conns: conns.remove(ws) if not conns: self.rooms.pop(room_id, None) async def broadcast(self, room_id: str, message: str): for ws in self.rooms.get(room_id, []): if ws.client_state WebSocketState.CONNECTED: await ws.send_text(message)这里的WebSocketState判断是必须的。客户端已经断开但还留在列表里时直接send会抛异常所以广播前先检查连接状态。实际项目里还要考虑send时发现异常后自动disconnect防止僵尸连接堆积。4.2 REST接口创建房间与获取房间列表房间元数据和连接集合分开管理REST接口只操作元数据连接状态查询借住manager。代码是这样的from starlette.applications import Starlette from starlette.routing import Route, WebSocketRoute from starlette.responses import JSONResponse from starlette.requests import Request room_metadata: Dict[str, dict] {} manager RoomManager() async def create_room(request: Request): payload await request.json() room_id payload.get(room_id) if not room_id or room_id in room_metadata: return JSONResponse({error: invalid room_id}, status_code400) room_metadata[room_id] {clients: 0} return JSONResponse({room_id: room_id}, status_code201) async def list_rooms(request: Request): result [] for room_id, meta in room_metadata.items(): online len(manager.rooms.get(room_id, [])) result.append({room_id: room_id, online: online}) return JSONResponse({rooms: result})在线人数我直接在查询时算不维护一个容易失真的计数器省心一些。如果你需要更精确的数据也可以把它维护成原子计数器但当前这种写法在单进程场景下完全够用。4.3 WebSocket端点消息广播与连接生命周期房间WebSocket端点是核心。连接进入时先校验房间是否存在不存在直接关闭并返回一个明确的错误codeasync def room_ws(websocket: WebSocket): room_id websocket.path_params[room_id] if room_id not in room_metadata: await websocket.close(code1008) return await manager.connect(room_id, websocket) room_metadata[room_id][clients] 1 try: while True: message await websocket.receive_text() await manager.broadcast(room_id, message) except WebSocketDisconnect: manager.disconnect(room_id, websocket) room_metadata[room_id][clients] - 1 except Exception: manager.disconnect(room_id, websocket) room_metadata[room_id][clients] - 1注意close(code1008)的含义是policy violation。这样客户端能明确感知到连接被拒绝而不是卡在一个不死不活的状态里。第二个except Exception是保底收尾防止一些非标准的断开行为让连接管理器留下脏数据。组装应用app Starlette(routes[ Route(/rooms, create_room, methods[POST]), Route(/rooms, list_rooms, methods[GET]), WebSocketRoute(/ws/{room_id}, room_ws), ])到这里一个多房间聊天服务的核心骨架已经有了。后面的验证环节会告诉你实际效果如何。4.4 用TestClient和真实浏览器验收测试时我直接用Starlette自带的TestClient它底层封装了httpx写起来非常顺手from starlette.testclient import TestClient client TestClient(app) resp client.post(/rooms, json{room_id: test-room}) assert resp.status_code 201 resp client.get(/rooms) assert test-room in resp.json()[rooms][0][room_id]真实运行用uvicorn启动uvicorn app:app --host 0.0.0.0 --port 8000然后打开浏览器控制台用原生WebSocket连接试试const ws new WebSocket(ws://localhost:8000/ws/test-room); ws.onopen () ws.send(hello from browser); ws.onmessage (e) console.log(received:, e.data);开两个标签页一个发消息另一个应该能立即收到。这个最小服务算是消息推送类项目的雏形后面要扩展成多进程广播只需要把RoomManager里的连接集合换成Redis的Pub/Sub通道即可。5. 常见问题与避坑实录5.1 lifespan事件压根不触发这个问题我遇到过很多次在Starlette(lifespan...)里写了启动逻辑结果启动时没执行。关键点是lifespan支持依赖ASGI服务器正确调用lifespan协议。如果你用的不是uvicorn这类完整支持ASGI lifespan的服务器或者测试环境里的TestClient没有走完整的触发流程事件就不会触发。建议用uvicorn跑服务测试时配合asgi-lifespan或pytest-asyncio手动触发。5.2 BaseHTTPMiddleware的性能和异常坑BaseHTTPMiddleware会把请求和响应都包装成stream在极高并发下性能比纯ASGI中间件差一截这是社区讨论过多次的老问题。另外在它内部捕获不到call_next里抛出的部分异常因为异常是在流式消费阶段才抛出的。解决方式要么把中间件写成纯ASGI类要么确保关键异常处理放进端点或最外层中间件。5.3 WebSocket断开后没清理连接这是我见过最多的坑。WebSocket连接断开后如果你没捕获WebSocketDisconnect或者在disconnect时没从管理器里移除对象后面的广播就会遇到一堆sent failed异常。解决方案就是前面实操里的那套异常处理块里统一disconnect广播前检查client_state同时管理器的remove操作要幂等同一个连接被remove两次也不会报错。5.4 后台任务的上下文隔离BackgroundTask里拿不到请求对象这一点新手最容易懵。因为请求在响应发送后已经销毁任务执行的环境和请求上下文完全不共享任何需要在任务里使用的数据都要在创建任务时显式传参。另外任务里要开数据库连接时千万别复用请求生命周期中创建的连接要重新获取。5.5 TestClient与事件循环的兼容问题直接跑TestClient一般没事但一旦你自己的代码里也用了asyncio事件循环就可能出现ProactorEventLoop不兼容之类的报错。我的经验是测试代码一律使用pytest-asyncio并且尽量不在TestClient所在线程里手动创建新事件循环保持单线程单循环。遇到websocket测试配合TestClient自带的websocket_connect上下文管理器就好。5.6 路径匹配顺序与404陷阱路由是逐个匹配的一旦前面的路由匹配了更宽的路径后面的具体路由就永远到不了。比如先写了/{room_id}再写/rooms那/rooms请求会先进第一个路由然后404。所以静态路径要写在动态路径之前这一点和所有框架的通用规则一致但Starlette因为路由写法太简洁很容易忽视。最后再聊点建议最后分享一点我自己的使用习惯。我用Starlette比较狠的那段时间主要是在做一个内部网关层前面接收所有流量后面转发到不同业务服务中间自己实现鉴权、限流和路由分发。当时没有现成的网关框架能满足定制需求用Starlette反而最省事。也正因为如此我把ASGI协议相关的坑差不多踩了一遍从lifespan到WebSocket连接管理都有深刻印象。有个经验值得强调不管最终选什么框架先把底层这层搞明白没有坏处。尤其当你用FastAPI时很多看似高级的能力其实都是Starlette打好的底子理解了它之后你的排查能力会上一个台阶。如果你只是想快速搭一个内部服务、做WebSocket转发或者单纯学异步Web开发Starlette都算是一条非常合适的路子。