- 使用 Starlette 创建 ASGI Web 应用
今天，我们大部分上网冲浪所依赖的网站，背后都是由各种各样的网页应用支撑着。如果你在一家有网络存在感的公司工作过，那么你很可能在职业生涯中曾参与过开发某个网页应用。在过去那种同步运行的 Python 世界里，这意味着你很可能用过 Flask、Bottle，或者极其流行的 Django 这些框架。除了最近几年的 Django 版本外，这些框架都并非为原生支持 asyncio 而设计。因此，当我们处理像数据库查询或调用外部 API 这类能并行化的工作时，除了多线程或多进程之外，没有太多选择可言。这就意味着，我们必须探索一些兼容 asyncio 模式的全新框架。
在这章中，我们将学习几个主流的、已准备好使用 asyncio 构建的 Web 框架。首先会通过我们已经熟悉的 aiohttp 框架，来构建一个异步的 RESTful API。接着深入理解一下所谓的“异步服务器网关接口”（ASGI）——它就是 WSGI 的异步版，是如今大多数 Web 应用的底层运行机制。我们会结合 Starlette 框架，用 ASGI 实现一个支持 WebSocket 的简单 REST API。最后，还会看看如何在 Django 中使用异步视图。毕竟性能始终是扩展应用时要考虑的核心问题，所以我们也会通过压力测试工具来实测这些方案的性能表现。
9.1 使用 aiohttp 创建 REST API
前面我们曾用 aiohttp 作为客户端，对多个网页应用发起成千上万的并发请求。但其实，aiohttp 不仅是个好用的客户端，还能用来搭建基于 asyncio 的异步服务器，真可谓是“一鱼多吃”。
REST，全称是“表示性状态转移”（Representational State Transfer），这可是现代网页开发里的头号热门范式，尤其在配合 React、Vue 等单页应用框架时更是一绝。它让我们能够以一种无状态、结构化的方式设计自己的网页接口，完全独立于前端的技术栈。一个合格的 REST API 应该可以无缝对接从手机到浏览器等各种客户端，而所需做的仅仅是调整前端对数据的展示方式。
说到 REST，关键就在于一个概念：资源（resource）。通常来说，任何名词化的实体都可以是一个资源。比如客户、商品或账户，都可以是属于 REST 体系的资源。刚才提到的那些例子，描述的是单个客户或商品这类单一资源。但“资源”也可以是一组集合，比如“顾客列表”或“商品列表”，通过唯一标识符就可以访问其中的某一项。而单项资源本身还可能拥有子资源。举个例子，一位顾客可能会有一份“最爱商品”的列表，这个“最爱列表”就成为这位顾客的子资源了。
customerscustomers/{id}customers/{id}/favorites
这里一共定义了三个端点。第一个 customers 是个集合端点，指向所有顾客的列表。消费者应该预期返回一个客户列表（这个列表可能分页，因为数据量可能很大）。第二个端点指向某个具体的顾客，需要传入一个 {id} 参数。如果客户编号是整数，那么请求 customers/1 就会返回编号为 1 的那位客户的详细信息。第三个端点则是子资源的例子：一个顾客可以拥有一串“喜爱的商品”，这就构成了它的子集合。访问 customers/1/favorites 就能拿到编号为 1 的顾客的“最爱商品”列表。
今后我们设计的所有 REST API 都将以返回 JSON 格式数据为目标，这已是行业标准。当然，理论上也可以根据需要选择其他格式。有些 REST API 还支持通过 HTTP 头部的内容协商（content negotiation）来返回多种数据形式。
当然，要完整掌握所有关于 REST 的细节，光靠这一章显然不够。想深入了解的话，建议去看其创造者的博士论文，里面内容非常透彻。地址在这里：http://mng.bz/1jAg
咱们现在动手，先搞个最简单的“欢迎来到世界”风格的 API。就拿一个 GET 请求做个示范，这个接口会返回当前的时间和日期信息。我们把这个接口叫 /time，目标是返回月、日和此刻时间。
aiohttp 在 web 模块里提供了完整的服务器功能。导入之后，你可以用 RouteTableDef 来定义路由（也就是在 aiohttp 里所谓的“路径”）。RouteTableDef 提供了一个装饰器，允许你指定请求类型（GET、POST 等）以及路径名称。然后用这个装饰器去修饰你的协程函数，当客户端访问该路径时，这个协程就会自动执行。在协程内部，你可以在进行完所需业务逻辑后，直接返回数据给客户端。
但请注意，只创建这些“蓝图”是没用的，你还得真正启动应用服务才行。具体做法是：先创建一个 Application 实例，把刚才定义好的路由注册进去，然后运行这个应用。
from aiohttp import webfrom datetime import datetimefrom aiohttp.web_request import Requestfrom aiohttp.web_response import Responseroutes = web.RouteTableDef()@routes.get('/time') <span class="fm-combinumeral">❶</span>async def time(request: Request) -> Response: today = datetime.today() result = { 'month': today.month, 'day': today.day, 'time': str(today.time()) } return web.json_response(result) <span class="fm-combinumeral">❷</span>app = web.Application() <span class="fm-combinumeral">❸</span>app.add_routes(routes)web.run_app(app)
❶ 创建 的 接口，客户端调用时就会触发 协程。❷ 把字典结果转为 JSON 响应返回。❸ 创建应用，注册路由，并启动服务。
代码解读：我们先定义了一个 time 接口。@routes.get('/time') 表示当客户端向 /time 发起 HTTP GET 请求时，对应的协程将被触发。在 time 协程中，我们获取当前的月份、日期和时间，打包进字典。然后调用 web.json_response，它会把字典序列化成 JSON 格式，同时设置响应头，例如把状态码设为 200，内容类型（Content-Type）设为 'application/json'。
接下来创建并启动应用：先实例化一个 Application，再调用 add_routes 注册所有路径。最后用 run_app 启动服务器，默认会在本地的 8080 端口启动。
运行后，你可以打开浏览器输入 localhost:8080/time，或者用命令行工具如 curl 测试。我们来试下 curl -i localhost:8080/time，你应该能看到类似下面的输出：
HTTP/1.1 200 OKContent-Type: application/json; charset=utf-8Content-Length: 51Date: Mon, 23 Nov 2020 16:35:32 GMTServer: Python/3.9 aiohttp/3.6.2{"month": 11, "day": 23, "time": "11:35:32.033271"}
瞧！咱第一个基于 aiohttp 的接口就这么成功诞生啦！
有个小细节可能你已经注意到了：我们的 time 协程里只接收一个叫 request 的参数。虽然这次没用上，但它未来会派上大用场。这个对象封装了客户端发来的所有请求信息，包括请求体、查询参数等等。想看请求头？只要在 time 协程里加一句 print(request.headers)，就能看到类似这样的输出：
<CIMultiDictProxy('Host': 'localhost:8080', 'User-Agent': 'curl/7.64.1', 'Accept': '*/*')>
刚才那个时间接口虽说展示了基础原理，但绝大多数实际的网页应用都远比这复杂得多。我们一般都需要连接数据库（如 Postgres、Redis），可能还要与其他 REST API 交互，比如查询或更新供应商提供的接口数据。
为了演示真实场景，我们现在基于第 5 章的电商商城数据库，来构建一个完整的、可操作的 REST API。目标是实现“读取现有商品”和“新增商品”两个功能。
第一步，我们需要建立与数据库的连接。考虑到应用会面临大量并发用户，用连接池而不是单个连接才是明智之举。那问题来了：这个连接池该放在哪儿，才能让所有路由都能轻松访问？
答案藏在 aiohttp 的设计哲学里：Application 类其实就是一个可以存取任意数据的字典。比如，如果你想让所有路由共享一份数据，可以直接这样存：
app = web.Application()app['shared_dict'] = {'key': 'value'}
那么想要获取这份数据时，只需执行 app['shared_dict'] 就行了。
另一个关键问题是：怎么在各个路由中获取 Application 本身的引用？虽然可以全局挂一个 app 变量，但 aiohttp 提供了更好的方案 —— 通过 Request 对象。每个进入的请求，都自带一个指向 Application 本身的 app 字段，让你能随时访问到共享数据。比如，你要获取刚才存储的字典，代码可以写成这样：
@routes.get('/')async def get_data(request: Request) -> Response: shared_data = request.app['shared_dict'] return web.json_response(shared_data)
接下来，我们就用这套机制来存放和管理数据库连接池。
现在，我们就要决定最佳时机来创建连接池。不能在初始化 Application 时直接做，因为那时还不是协程环境，无法使用 await。
aiohttp 提供了一个贴心的解决方案：在 Application 实例上绑定了信号处理器 on_startup，专门用来处理启动时的初始化任务。你完全可以把它想象成一组启动时要跑的协程队列。想添加一个协程，就调用 app.on_startup.append(coroutine)。每个加入这个列表的协程，都会接收到一个 Application 实例作为参数。我们可以在这个协程里创建好连接池，然后存进 app 实例。
另外，还要考虑关闭应用时的清理工作。必须主动关闭数据库连接，否则可能留下“僵尸连接”，对数据库造成不必要的压力。aiohttp 也贴心地提供了 on_cleanup 信号处理器，应用关闭时它会自动执行里面的协程。行为和 on_startup 一模一样，同样是通过 append 加入你想要清理的任务。
好了，把这些零散的部分拼起来，我们就能打造一个带有数据库连接池的完整应用。为了验证效果，先实现一个接口，用来查出数据库里所有的品牌信息。这是一个 GET 接口，路径名为 /brands。
import asyncpgfrom aiohttp import webfrom aiohttp.web_app import Applicationfrom aiohttp.web_request import Requestfrom aiohttp.web_response import Responsefrom asyncpg import Recordfrom asyncpg.pool import Poolfrom typing import List, Dictroutes = web.RouteTableDef()DB_KEY = 'database'async def create_database_pool(app: Application): <spanclass="fm-combinumeral">❶</span> print('Creating 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[DB_KEY] = poolasync def destroy_database_pool(app: Application): <spanclass="fm-combinumeral">❷</span> print('Destroying database pool.') pool: Pool = app[DB_KEY] await pool.close()@routes.get('/brands')async def brands(request: Request) -> Response: <spanclass="fm-combinumeral">❸</span> connection: Pool = request.app[DB_KEY] 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 web.json_response(result_as_dict)app = web.Application()app.on_startup.append(create_database_pool) <span class="fm-combinumeral">❹</span>app.on_cleanup.append(destroy_database_pool)app.add_routes(routes)web.run_app(app)
❶ 创建数据库连接池，并存进应用实例。❷ 从应用实例中销毁连接池。❸ 查询所有品牌并返回结果。❹ 将创建和销毁连接池的协程分别加入启动和关闭流程。
解释如下：我们定义了两个协程用于创建和销毁连接池。create_database_pool 里创建连接池，并以 DB_KEY 为键存进 app。destroy_database_pool 则取出连接池并等待其关闭。在启动应用时，我们分别将这两个协程添加进 on_startup 和 on_cleanup 信号处理器。
接着定义 brands 路由：先从请求中拿到数据库连接池，然后执行查询语句获取全部品牌数据。随后遍历每一条记录，转换成字典形式，因为 aiohttp 无法直接序列化 asyncpg.Record 类型的对象。当你运行这个应用后，打开浏览器访问 localhost:8080/brands，应该能看到形如下面的响应：
[{"brand_id": 1, "brand_name": "his"}, {"brand_id": 2, "brand_name": "he"}, {"brand_id": 3, "brand_name": "at"}]
搞定！这是我们第一个真正的集合作为资源的 RESTful 接口。接下来，我们尝试创建更多针对“单一资源”的接口，比如根据唯一编号获取某个商品，或者创建一个新商品。
让我们从获取商品开始。这个接口需要接受一个整数型的 id 参数，比如获取编号为 1 的商品，就请求 /products/1。如何做到在路径中动态传递参数呢？aiohttp 支持用大括号 {} 包裹参数名，所以这个接口路径应该是 /products/{id}。一旦如此定义，请求的 match_info 字典里就会包含这个参数值。你可以通过 request.match_info['id'] 获取它，但返回的是字符串形式。
由于客户端可能传入非法字符（比如字母），我们需要加入错误处理。此外，还有可能是用户请求一个不存在的 id，我们也得妥善处理“未找到”情况。对于这类错误，我们返回标准的 400 错误码表示客户端请求错误；如果是请求不存在的数据，则返回 404。aiohttp 提供了一整套预定义的异常类对应不同的状态码，我们只需要抛出它们，客户端自然就能收到正确的状态码。
import asyncpgfrom aiohttp import webfrom aiohttp.web_app import Applicationfrom aiohttp.web_request import Requestfrom aiohttp.web_response import Responsefrom asyncpg import Recordfrom asyncpg.pool import Poolroutes = web.RouteTableDef()DB_KEY = 'database'@routes.get('/products/{id}')async def get_product(request: Request) -> Response: try: str_id = request.match_info['id'] <span class="fm-combinumeral">❶</span> product_id = int(str_id) query = \ """ SELECT product_id, product_name, brand_id FROM product WHERE product_id = $1 """ connection: Pool = request.app[DB_KEY] result: Record = await connection.fetchrow(query, product_id) <span class="fm-combinumeral">❷</span> if result is not None: <span class="fm-combinumeral">❸</span> return web.json_response(dict(result)) else: raise web.HTTPNotFound() except ValueError: raise web.HTTPBadRequest()async def create_database_pool(app: Application): print('Creating 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[DB_KEY] = poolasync def destroy_database_pool(app: Application): print('Destroying database pool.') pool: Pool = app[DB_KEY] await pool.close()app = web.Application()app.on_startup.append(create_database_pool)app.on_cleanup.append(destroy_database_pool)app.add_routes(routes)web.run_app(app)
❶ 从 URL 中获取 参数。❷ 执行单条商品查询。❸ 如果查询成功，转换为 JSON 返回；否则返回“404 未找到”。
接着，我们来实现创建商品的接口。客户端会通过请求体发送一个包含商品信息的 JSON，然后我们在服务器端解析并插入数据库。这部分也得做些校验，确保传入的 JSON 格式正确。如果不对，就得返回 400 错误。
import asyncpgfrom aiohttp import webfrom aiohttp.web_app import Applicationfrom aiohttp.web_request import Requestfrom aiohttp.web_response import Responsefrom chapter_09.listing_9_2 import create_database_pool, destroy_database_poolroutes = web.RouteTableDef()DB_KEY = 'database'@routes.post('/product')async def create_product(request: Request) -> Response: PRODUCT_NAME = 'product_name' BRAND_ID = 'brand_id' if not request.can_read_body: raise web.HTTPBadRequest() body = await request.json() if PRODUCT_NAME in body and BRAND_ID in body: db = request.app[DB_KEY] await db.execute('''INSERT INTO product(product_id, product_name, brand_id) VALUES(DEFAULT, $1, $2)''', body[PRODUCT_NAME], int(body[BRAND_ID])) return web.Response(status=201) else: raise web.HTTPBadRequest()app = web.Application()app.on_startup.append(create_database_pool)app.on_cleanup.append(destroy_database_pool)app.add_routes(routes)web.run_app(app)
代码说明：我们首先检查请求是否包含可读的请求体，如果没有就立即返回坏请求。接着用 await request.json() 方法获取整个请求体，并转化为字典。为什么它是协程？因为如果请求体特别大，读取过程可能会被缓冲，耗时较久。如果我们不用 await 等待，就会阻塞整个协程处理流程。用 await 等待数据完全加载完毕后，我们再把数据插入数据库，最后返回一个 201 Created 的状态码，告诉客户端成功了。
用 curl 测试一下，你大概可以运行类似下面这条命令来插入一条商品：
curl -i -d '{"product_name":"product_name", "brand_id":1}' localhost:8080/productHTTP/1.1 201 CreatedContent-Length: 0Content-Type: application/octet-streamDate: Tue, 24 Nov 2020 13:27:44 GMTServer: Python/3.9 aiohttp/3.6.2
你看，这么一顿操作下来，我们就完成了对数据库的操作。当然，生产环境里肯定得加上更健壮的错误处理（比如防止 brand_id 是字符串的情况，或修复畸形的 JSON），但这已经清楚地展示了如何处理 postdata 并写入数据库。
9.1.4 aiohttp 与 Flask 的对比
当你用 aiohttp 和 asyncio 框架开发时，最大的优势之一就是能自由使用 asyncpg 这类异步库。但说真的，除了能直接使用异步库之外，像 aiohttp 这种框架相比 Flask 这类同步框架，还有什么额外好处吗？
尽管结果很大程度上取决于服务器配置、数据库硬件等条件，但总体而言，基于 asyncio 构建的应用通常能以更少的资源实现更高的吞吐量。在同步框架中，每个请求处理器都是从头到尾“一口气”执行完才退出；而在异步框架里，当遇到 await 表达式时，程序暂停，系统能把空闲时间分配给其他任务，从而显著提升效率。
为了直观感受这一点，我们来手撸一个 Flask 版本的“品牌查询”接口。假设你对 Flask 还不熟也没关系，基本概念还是好懂的。先安装依赖：
pip install -Iv flask==2.0.1pip install -Iv psycopg2==2.9.1
如果是用 psycopg 安装失败，可能是因为缺少编译依赖，这时你得装上 PostgreSQL 工具、OpenSSL 或其他库。百度一下报错信息，一般都有解决办法。
现在，我们来实现接口：先连接数据库，然后在请求处理器里复用之前写的查询语句，把结果转成 JSON 数组返回。
from flask import Flask, jsonifyimport psycopg2app = Flask(__name__)conn_info = "dbname=products user=postgres password=password host=127.0.0.1"db = psycopg2.connect(conn_info)@app.route('/brands')def brands(): cur = db.cursor() cur.execute('SELECT brand_id, brand_name FROM brand') rows = cur.fetchall() cur.close() return jsonify([{'brand_id': row[0], 'brand_name': row[1]} for row in rows])
准备工作做好后，我们需要运行应用。Flask 自带的开发服务器虽方便，但不适合生产环境，且只能单进程运行，意味着同一时间只能处理一个请求。要公平比较，得用专业的生产级服务器。这里我们选 Gunicorn。
pip install -Iv gunicorn==20.1.0
我们是在一台 8 核机子上测试，于是启动 8 个工作进程：
gunicorn -w 8 chapter_09.listing_9_5:app
你会看到以下输出，表明 8 个工作进程已经成功启动：
[2020-11-24 09:53:39 -0500] [16454] [INFO] Starting gunicorn 20.0.4[2020-11-24 09:53:39 -0500] [16454] [INFO] Listening at: http://127.0.0.1:8000 (16454)[2020-11-24 09:53:39 -0500] [16454] [INFO] Using worker: sync[2020-11-24 09:53:39 -0500] [16458] [INFO] Booting worker with pid: 16458...
这意味着我们打开了 8 个数据库连接，理论上能同时处理 8 个请求。现在，我们要找一个压测工具来做对比。wrk 是个绝佳的命令行压测神器，任何压测工具如 Apache Bench 都行。官网安装教程(https://github.com/wg/wrk) 上有详细介绍。
现在启动 30 秒的压测，模拟 200 个并发客户端同时疯狂点击：
Running 30s test @ http://localhost:8000/brands 1 threads and 200 connections 16534 requests in 30.02s, 61.32MB read Socket errors: connect 0, read 1533, write 276, timeout 0Requests/sec: 550.82Transfer/sec: 2.04MB
平均响应约 550 次/秒，挺不错。接下来用 aiohttp 重跑一遍：
Running 30s test @ http://localhost:8080/brands 1 threads and 200 connections 46774 requests in 30.01s, 191.45MB readRequests/sec: 1558.46Transfer/sec: 6.38MB
aiohttp 成功达到了 1558 次/秒，几乎是 Flask 的三倍！最关键的是，这只是一个进程完成的，而 Flask 为了扛住这点流量，竟需要整整 八个进程 才能做到三分之一的性能！你甚至还能进一步优化——比如在 aiohttp 前面加一层 NGINX 反向代理，并增加更多工作进程，性能还能再上一个台阶。
从此我们知道，借助 aiohttp 构建基于数据库的异步应用，是多么高效的一件事。但要注意，aiohttp 有点与众不同，它本身就是一个独立的服务器，完全不依赖 WSGI，也不遵循这种规范，可以自成一体。而大多数框架（比如我们后面用到的 Django）就没有这么自由。
