FastPySGI 基准测试实现解析：基于 libuv 与 llhttp 的 Python WSGI/ASGI 服务器

• 性能测试
• 测试
• 后端
• 开发者工具

【免费下载链接】FrameworkBenchmarks

Source for the TechEmpower Framework Benchmarks project

项目地址：
https://gitcode.com/gh_mirrors/fr/FrameworkBenchmarks

点击查看 免费下载

本指南以 FrameworkBenchmarks 仓库中 FastPySGI 测试目录 为蓝本，系统讲解 FastPySGI 这一基于 libuv 事件循环的 Python 3 异步服务器的技术构成，以及它在 TechEmpower 基准测试套件中的 WSGI、ASGI 双实现方式。读完本文，你将掌握 FastPySGI 的架构原理、app.py 与 app-asgi.py 两套测试入口的完整路由实现、benchmark_config.json 与 config.toml 中测试档位的配置方法，以及如何借助 tfb –mode verify –test fastpysgi 在本地验证该测试。

FastPySGI 是什么

FastPySGI 是一个面向 Python 3 的 asyncio 高性能服务器，其核心设计是借助 C 库将最繁重的网络与解析工作下沉到原生层：

• libuv：多平台 C 库，提供基于事件循环的异步 I/O 支持，这也是 Node.js 同款的事件循环基石；
• llhttp：Node.js 社区维护的 HTTP 解析库，以解析速度快著称；
• WSGI/ASGI 消费者接口：用于与应用层交互，使 FastPySGI 同时具备接入经典 WSGI 应用与新型 ASGI 应用的能力。

在 fastpysgi/README.md 中，作者将其定位为 "lightning fast asyncio server"（基于 libuv 的极速 asyncio 服务器）。值得说明的是，本项目中的实现还参考了同为极简设计的 FastWSGI（README 的 Resources 一节将其列为关联资源），两者均走"单文件、零框架依赖"的实现路线。

测试目录结构与配置映射

在 frameworks/Python/fastpysgi 目录下，测试实现被组织为如下文件：

README 明确指出：所有测试实现都位于单个文件 app.py 中，并给出两个测试路径映射——JSON Serialization 对应 /json，Plaintext 对应 /plaintext。实际上该目录下还存在功能更完整的 ASGI 版本 app-asgi.py，两个入口各自被 benchmark_config.json 中的不同档位引用，这一点在后续章节详细展开。

WSGI 实现：极简的单文件路由

app.py 是 FastPySGI 的 WSGI 实现，全文件仅约 60 行，遵循经典 WSGI 签名 app(environ, start_response)。核心代码 frameworks/Python/fastpysgi/app.py#L10-L26 展示了一条典型的 WSGI 处理链路：

def app(environ, start_response):
path = environ["PATH_INFO"]
headers = [ ]

if path == "/json":
headers.append( ('Content-Type', 'application/json') )
start_response('200 OK', headers)
return [ jsonify( {"message":"Hello, World!"} ).encode('utf8') ]

if path == "/plaintext":
headers.append( ('Content-Type', 'text/plain') )
start_response('200 OK', headers)
return [ b'Hello, World!' ]

start_response('400 Bad Request', headers)
return [ b'' ]

几个值得注意的实现细节：

• JSON 序列化优先使用 ujson：app.py 顶部通过 try: from ujson import dumps as jsonify except: from json import dumps as jsonify 做双路回退（app.py#L4-L7），即在具备 ujson 时用 C 扩展加速序列化，缺失时回退到标准库，这与 requirements.txt 中锁定 ujson==5.12.0 的版本一致；
• 路径分发完全由 PATH_INFO 驱动：不引入任何路由框架，/json 返回 application/json 响应体 {"message":"Hello, World!"}，/plaintext 返回 text/plain 响应体 Hello, World!，其余路径一律回 400 Bad Request；
• 注释中的 Server 头默认关闭：代码中保留了 #headers = [ ('Server', 'FastPySGI') ] 注释，说明作者默认省略自定义 Server 响应头以压缩响应字节数。

多进程启动模型

__main__ 块（app.py#L29-L55）展示了 FastPySGI 在基准测试中的典型部署方式——利用 os.fork() 按 CPU 核数创建多个 worker 进程：

workers = int(multiprocessing.cpu_count())

host = '0.0.0.0'
port = 3000

def run_app():
fastpysgi.server.backlog = 4096
fastpysgi.run(app, host, port, loglevel=2)

def create_fork():
n = os.fork()
if not n > 0:
run_app()

# fork limiting the cpu count – 1
for i in range(1, workers):
create_fork()

run_app() # run app on the main process too

关键参数含义：

• port = 3000：监听端口，与 benchmark_config.json 中 "port": 3000 严格对应；
• host = '0.0.0.0'：绑定所有网卡接口，便于容器内被 wrk 压测访问；
• fastpysgi.server.backlog = 4096：调高 listen backlog 队列长度，在高并发下减少内核 accept 队列丢包；
• loglevel=2：日志级别配置，测试运行时用于输出诊断信息；
• 启动策略是"主进程 + N-1 个 fork 子进程"，每个子进程各自调用 fastpysgi.run() 运行同一应用实例；当环境变量 TRAVIS=true 时（CI 场景）强制只启动 2 个 worker，避免在资源受限环境打满 CPU。

ASGI 实现：全量测试路径与异步数据库访问

app-asgi.py 是功能更完整的 ASGI 实现，README 虽未逐条列出，但它覆盖了 TechEmpower 套件的全部七类测试。它遵循 ASGI 三方协议 app(scope, receive, send)，并利用 lifespan 协议在进程启动阶段初始化数据库连接池（app-asgi.py#L232-L247）。

数据库连接池

db_setup()（app-asgi.py#L28-L39）基于 asyncpg 建立到 Postgres 的连接池：

db_pool = await asyncpg.create_pool(
user=os.getenv('PGUSER', 'benchmarkdbuser'),
password=os.getenv('PGPASS', 'benchmarkdbpass'),
database='hello_world',
host='tfb-database',
port=5432,
min_size=PG_POOL_SIZE,
max_size=PG_POOL_SIZE,
connection_class=NoResetConnection,
)

其中 PG_POOL_SIZE = 4 为固定连接池大小；host='tfb-database' 是套件内部约定、由 Docker Compose 注入的数据库容器主机名；连接参数 benchmarkdbuser/benchmarkdbpass 与 toolset/databases/postgres/postgres.py 中校验端使用的账号密码完全一致。NoResetConnection 继承自 asyncpg.Connection 并覆写 get_reset_query() 返回空串，用于跳过连接归还时的 RESET 查询，省去不必要的往返开销。

路由表与七类测试

路由表（app-asgi.py#L221-L229）将路径映射到协程处理器：

routes = {
'/json': json_serialization,
'/db': single_database_query,
'/queries': multiple_database_queries,
'/fortunes': fortunes,
'/updates': database_updates,
'/cached-queries': cached_queries,
'/plaintext': plaintext,
}

• /json 与 /plaintext：与 WSGI 版功能一致，但改由 ASGI 消息结构发送——先 send 一个 http.response.start 消息声明状态码与 Content-Type 头，再 send 一个 http.response.body 消息携带 more_body: False 的响应体；
• /db 单条数据库查询：random.randint(1, 10000) 随机取 id，fetchval(READ_ROW_SQL, row_id) 读取 world 表的 randomnumber，返回 {"id":…, "randomNumber":…}（app-asgi.py#L92-L107）；
• /queries 多条查询：通过 get_num_queries(scope) 解析 queries 查询参数，用 random.sample(range(1, 10000), num_queries) 生成互不重复的 id 集，再以 fetchmany 一次取回；get_num_queries（app-asgi.py#L69-L79）对参数做了 1～500 的钳制——小于 1 视为 1，大于 500 钳为 500，非法输入回退为 1，这与 toolset/test_types/verifications.py#L329-L349 中 verify_query_cases 对 queries 参数 0、负数、非数字、501、空串等用例的校验预期一致；
• /fortunes：从数据库拉取 Fortune 表全部数据，追加一行 [0, "Additional fortune added at request time."] 后按 id 排序，交给 templates/fortune.html 中的 Jinja2 模板渲染，模板对 fortune[1] 做了 |e HTML 转义以处理 XSS 用例（数据库第 11 条记录即包含 <script> 标签的测试数据，见 create-postgres.sql）；
• /updates 数据库更新：随机生成 id 与随机数，先 executemany(READ_ROW_SQL, …) 逐条读取、再 executemany(WRITE_ROW_SQL, updates) 逐条更新，返回更新后的 world 列表；
• /cached-queries 缓存查询：这是 FastPySGI 实现的一个亮点——通过 asyncache 的 @cached 装饰器配合 cachetools.keys.hashkey 按 id 做异步记忆化（app-asgi.py#L175-L181），并使用 db_conn.prepare(READ_ROW_SQL) 预编译语句减少 SQL 解析开销，参数名改为 count。

所有 ASGI 响应消息均预定义常量（JSON_RESPONSE、HTML_RESPONSE、PLAINTEXT_RESPONSE），复用响应头以省去重复构造。

ASGI 入口的启动方式

与 WSGI 版相同的 fork 多进程模型，但每个 worker 启动前先 asyncio.get_event_loop().run_until_complete(db_setup()) 完成连接池初始化（app-asgi.py#L251-L281），随后同样设置 backlog = 4096 并调用 fastpysgi.run(app, host, port, loglevel=2)。

测试档位配置：benchmark_config.json 与 config.toml

benchmark_config.json 定义了 default（WSGI） 与 asgi（ASGI） 两个测试档位，二者的关键差异如下表：

versus 字段值得关注：WSGI 档位将同为 Python 3 WSGI 服务器的 wsgi 作为对照基线，ASGI 档位则将 uvicorn 作为对照基线，表明 FastPySGI 的目标是分别与两类服务器同台竞技。maintainers 字段登记了维护者 remittor（即 FastPySGI 上游作者）。

与 JSON 配置等价的 config.toml 采用 TOML 语法，分为 [framework]、[main]、[asgi] 三节，其中 [main] 对应 WSGI 档位（platform = "WSGI"、versus = "wsgi"），[asgi] 对应 ASGI 档位（platform = "ASGI"、versus = "uvicorn"）。两份配置同时保留，供测试套件按需选用。

容器化构建与依赖锁定

两个 Dockerfile 均基于 python:3.11-bullseye 镜像，流程完全一致，仅启动命令不同：

• fastpysgi.dockerfile：COPY requirements.txt ./ 后先 pip3 install -U pip，再 pip3 install –no-cache-dir -r requirements.txt 无缓存安装依赖，随后 COPY . . 拷贝源码，EXPOSE 3000 声明端口，最后 CMD python ./app.py 启动 WSGI 应用；
• fastpysgi-asgi.dockerfile：除 CMD python ./app-asgi.py 外其余步骤完全相同。

requirements.txt 锁定了六个依赖及其精确版本：

fastpysgi==0.2
ujson==5.12.0
asyncpg==0.30.0
jinja2==3.1.6
cachetools==5.3.0
asyncache==0.3.1

其中 fastpysgi==0.2 是核心服务器本体；asyncpg 服务于 ASGI 档位的 Postgres 连接池；jinja2 服务于 Fortune 模板渲染；cachetools 与 asyncache 服务于 /cached-queries 的异步缓存。

测试如何被套件校验

FastPySGI 的每个测试路径都要经过 toolset/test_types 下对应测试类型的严格校验，这些校验逻辑构成了"实现必须合规"的边界：

• JSON 测试（toolset/test_types/json/json.py）：响应体必须能反序列化为 JSON，且包含值为 hello, world!（大小写不敏感）的 message 键，同时校验 200 OK 状态码与 Content-Type: application/json 响应头（正则 ^application/json(; ?charset=(UTF|utf)-8)?$，见 verifications.py#L104-L117）；
• Plaintext 测试（toolset/test_types/plaintext/plaintext.py）：响应体必须包含 hello, world!（大小写不敏感），并警告超出必要字节数；
• db / query / update / cached-query 测试：均依赖 verify_randomnumber_object / verify_randomnumber_list 校验 id、randomNumber 键与数值范围（1～10000），query 与 update 还会通过 verify_query_cases 以 ('2','fail'), ('0','fail'), ('foo','fail'), ('501','warn'), ('','fail') 五组用例验证参数钳制行为（query.py#L28-L30）；
• Fortune 测试（toolset/test_types/fortune/fortune.py）：用 FortuneHTMLParser 解析 HTML 响应并与合法输出逐行 diff，同时校验查询次数与读取行数（12 行 × 请求数）；
• 查询次数统计：verify_queries_count（verifications.py#L420-L469）借助 Postgres 的 pg_stat_statements 扩展统计实际执行的 SQL 次数与读取行数（对应 postgres.py#L59-L68 中的 get_queries/get_rows/get_rows_updated），防止框架"作弊"绕过数据库读取。

数据库侧的数据源由 create-postgres.sql 提供：World 表预置 10000 行随机数据（SELECT x.id, least(floor(random() * 10000 + 1), 10000) FROM generate_series(1,10000)），Fortune 表预置 12 条固定语录。这正是 random.randint(1, 10000) 取值范围与 worlds 校验区间 1～10000 的来源。

本地验证与运行

若要验证 FastPySGI 测试实现是否正确，可参考 toolset/scaffolding/README.md 中描述的标准流程，在仓库根目录执行：

$ tfb –mode verify –test fastpysgi

该命令会以 verify 模式运行套件，不采集性能数据，仅测试能否命中 benchmark_config.json 中声明的端点且响应内容正确。需要注意的适用前提：app-asgi.py 依赖 tfb-database 主机名与 benchmarkdbuser/benchmarkdbpass 凭据，因此 ASGI 档位的验证需要套件拉起 Postgres 数据库容器；而 WSGI 档位（/json、/plaintext）无数据库依赖，可以独立运行。手动本地运行则可直接执行：

python ./app.py # WSGI 版，监听 0.0.0.0:3000
python ./app-asgi.py # ASGI 版，需先保证 Postgres 可达

两个入口都支持通过环境变量 TRAVIS=true 将 worker 数钳制为 2，便于在 CI 环境快速冒烟测试。

小结

从这一份短短数行的 README 出发，可以完整还原 FastPySGI 测试的全貌：libuv + llhttp 提供原生异步 I/O 与 HTTP 解析，WSGI/ASGI 双消费者接口让同一服务器可以承载传统与异步两类 Python 应用。仓库中 app.py 的极简路由与 fork 多进程模型、app-asgi.py 的 asyncpg 连接池与异步缓存查询、benchmark_config.json 与 config.toml 的双档位配置、以及 toolset 中层层递进的响应校验与查询计数机制，共同构成了一条从"服务器原理"到"基准验证"的完整链路，也为在 FrameworkBenchmarks 中新增类似轻量级 Python 服务器的实现提供了可复用的参照模板。

赞

分享

• 性能测试
• 测试
• 后端
• 开发者工具

【免费下载链接】FrameworkBenchmarks

Source for the TechEmpower Framework Benchmarks project

项目地址：
https://gitcode.com/gh_mirrors/fr/FrameworkBenchmarks

点击查看 免费下载

相关推荐

• Falcon 在 TechEmpower Framework Benchmarks 中的 WSGI/ASGI 基准测试实现解析

  Falcon 在 TechEmpower Framework Benchmarks 中的 WSGI/ASGI 基准测试实现解析 本文以 frameworks/P

  性能测试
  测试
  后端
  开发者工具

• Falcon Web Framework 实战指南：基于 ASGI/WSGI 构建高性能 REST API 与微服务

  Falcon Web Framework 实战指南：基于 ASGI/WSGI 构建高性能 REST API 与微服务 Falcon 是一个以"极简、无魔法（no

  后端
  Web框架
  API设计

• gevent 基于 Unix Domain Socket 的 WSGI 服务器实战：unixsocket_server.py 深度解析

  gevent 基于 Unix Domain Socket 的 WSGI 服务器实战：unixsocket_server.py 深度解析 本指南以 gevent

  后端

上一篇：
BioGPT完整教程：生物医学AI的入门到精通指南

下一篇：
ADBKeyBoard完全指南：5分钟搞定Android自动化中文输入

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考