Python 并发与 asyncio 9.3 使用 Starlette 搭建 ASGI 应用

9.3 使用 Starlette 搭建 ASGI 应用

Starlette，是由 Encode 团队（Uvicorn 和 Django REST framework 的作者）打造的一个小巧灵活的 纯正 的 ASGI 框架。它以高性能、支持 WebSocket 和诸多强大功能著称。它的文档请戳：https://www.starlette.io/(https://www.starlette.io/)。

来实践一下，我们如何用它来快速搭建一个支持 REST 接口和 WebSocket 的应用。先安装：

pip install -Iv starlette==0.15.0

9.3.1 使用 Starlette 构建一个 REST 接口

我们先来重新实现之前的“品牌查询”接口。通过创建 Starlette 类的实例来构造应用，它有几个关键参数我们关心：routes（路由列表）和 on_startup / on_shutdown（启动和关闭时要运行的协程）。Route 对象就是把路径字符串（如 /brands）映射到协程或可调用对象。

和 aiohttp 很像，这个协程也只有一个 request 参数，返回一个响应。区别在于数据共享方式：Starlette 应用实例也有一个 state 属性，我们把连接池存进去。

列表 9.8 一个 Starlette 的品牌接口

import asyncpgfrom asyncpg import Recordfrom asyncpg.pool import Poolfrom starlette.applications import Starlettefrom starlette.requests import Requestfrom starlette.responses import JSONResponse, Responsefrom starlette.routing import Routefrom typing import List, Dictasync def create_database_pool():    pool: Pool = await asyncpg.create_pool(host='127.0.0.1',                                           port=5432,                                           user='postgres',                                           password='password',                                           database='products',                                           min_size=6,                                           max_size=6)    app.state.DB = poolasync def destroy_database_pool():    pool = app.state.DB    await pool.close()async def brands(request: Request) -> Response:    connection: Pool = request.app.state.DB    brand_query = 'SELECT brand_id, brand_name FROM brand'    results: List[Record] = await connection.fetch(brand_query)    result_as_dict: List[Dict] = [dict(brand) for brand in results]    return JSONResponse(result_as_dict)app = Starlette(routes=[Route('/brands', brands)],                on_startup=[create_database_pool],                on_shutdown=[destroy_database_pool])

应用搭好后，我们用 Uvicorn 启动它，同样开启 8 个工作进程：

uvicorn --workers 8 --log-level error chapter_09.listing_9_8:app

访问 http://localhost:8000/brands，应该能看到品牌数据的列表。

再跑一次压测，结果令人震惊：

Running 30s test @ http://localhost:8000/brands  1 threads and 200 connectionsRequests/sec:   4365.37Transfer/sec:   16.07MB

每秒超过 4300 次请求！这不仅完爆了 Flask，甚至超过了前面的 aiohttp（那次我们只用了单个 worker）。之所以如此，是因为异步框架能充分利用高并发下的等待时间。虽然是个非完全公平的对比（要是 aiohttp 也开了 8 个 worker + NGINX，数字会更接近），但足以证明 异步架构的巨大潜力。

9.3.2 使用 Starlette 处理 WebSocket

在传统的网页请求中，客户端发个请求，服务器回个响应，大功告成。但如果想实现“实时数据更新”，比如动态显示当前在线人数，该怎么办？

一种常见思路是：用前端的 JavaScript 定期轮询一个接口，问“现在有多少人在线？”每隔几秒就刷新一次页面。看似可行，实则是个“笨招”。每一次轮询，无论数据有没有变，都会在服务器上白白浪费一次请求和响应的开销。如果数据长期不变，那就纯粹是“打水漂”，徒增服务器压力。

这就好比车里孩子不停问：“到了吗？”——听着烦不说，根本没意义。

相比之下，WebSocket 是一剂良方。一旦建立起一个持久连接，客户端和服务端就可以在同一条“通道”上自由收发数据，无需每次都走完整的请求-响应流程。而且它支持双向通信。

回到在线人数的例子：客户端一连上，服务端就立刻知道有人连接，它再也不用等你频繁发问了。每当有新人进来，服务端直接推送更新；你关闭窗口，它也知道谁走了。看下图就很明白了：

图 9.1：轮询对比与 WebSocket

这正是星跳板（Starlette）的强项。它内建了非常易用的 WebSocket 支持。

要动手试试，首先装包：

pip install -Iv websockets==9.1

我们的目标是：创建一个简单的接口，当多个客户端连接上来时，实时显示有多少人在用。

计划是：维护一个内存中的客户端列表。有新用户进来，就加入列表并通知所有人；有用户离开，就把他的名字移除，并通知所有人数量变了。顺便加点容错处理：如果发送失败，说明客户端已断开，也得从列表里删掉。

在 Starlette 里，可以通过继承 WebSocketEndpoint 类来创建一个接口。这个类有三个必须实现的协程：

列表 9.9 一个 Starlette 的 WebSocket 接口

import asynciofrom starlette.applications import Starlettefrom starlette.endpoints import WebSocketEndpointfrom starlette.routing import WebSocketRouteclass UserCounter(WebSocketEndpoint):    encoding = 'text'    sockets = []    async def on_connect(self, websocket):                  <span class="fm-combinumeral">❶</span>        await websocket.accept()        UserCounter.sockets.append(websocket)        await self._send_count()    async def on_disconnect(self, websocket, close_code):   <span class="fm-combinumeral">❷</span>        UserCounter.sockets.remove(websocket)        await self._send_count()    async def on_receive(self, websocket, data):        pass    async def _send_count(self):                            <span class="fm-combinumeral">❸</span>        if len(UserCounter.sockets) > 0:            count_str = str(len(UserCounter.sockets))            task_to_socket = {asyncio.create_task(websocket.send_text(count_str)): websocket                              for websocket                              in UserCounter.sockets}            done, pending = await asyncio.wait(task_to_socket)            for task in done:                if task.exception() is not None:                    if task_to_socket[task] in UserCounter.sockets:                        UserCounter.sockets.remove(task_to_socket[task])app = Starlette(routes=[WebSocketRoute('/counter', UserCounter)])

❶ 客户端连接时，将其加入列表，并通知所有人更新计数。❷ 客户端断开时，从列表移除，并通知所有人更新计数。❸ 通知所有在线者，当前在线人数。如有异常，立即从列表删除。

然后，我们写一个前端页面来连接，接收消息并更新显示。

列表 9.10 使用这个接口

<!DOCTYPE html><html lang=""><head>    <title>Starlette Web Sockets</title>    <script>        document.addEventListener("DOMContentLoaded", () => {            let socket = new WebSocket("ws://localhost:8000/counter");            socket.onmessage = (event) => {                const counter = document.querySelector("#counter");                counter.textContent = event.data;            };        });    </script></head><body>    <span>Users online: </span>    <span id="counter"></span></body></html>

核心在 script 标签里。我们通过 WebSocket 对象连接，然后监听 onmessage 事件。一旦收到消息，就用 querySelector 找到 #counter 元素，并把收到的数据填进去。特别注意，我们是等到 DOMContentLoaded 事件触发后才运行脚本的，不然页面还没生成 #counter，那代码就会报错。

启动服务，用 uvicorn --workers 1 chapter_09.listing_9_9:app 运行，然后打开这个页面，你应该能看到屏幕上显示 1。如果你在多个标签页里都打开，你会发现所有页面的数字都在涨。关掉一个标签页，其他页面的数字马上减 1。

注意，我们这里只用了一个工作进程。这是因为空间内的 sockets 列表是内存共享的，若用多进程，每个进程都会有自己的副本，无法共享状态。部署到生产环境，就得用数据库之类持久化存储来统一管理状态。

到此为止，我们已经学会了如何用 aiohttp 与 Starlette 构建基于 asyncio 且兼具 REST 与 WebSocket 的现代应用。不过话说回来，这些框架即便再火，也无法与 Django 这个重量级选手相提并论——它在中文圈被誉为当之无愧的 “1000 斤猛兽” 之王。