不重写系统:用 PyO3 把 Python 性能热点逐步替换成 Rust

2026-09-10 24 预计阅读时间: 1 分钟
来源: infoq.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:9 分钟

当 Python 应用遇到性能瓶颈时,最危险的选择往往是“全部重写”。Lily Mara 提出的路径更务实:保留已经稳定运行的 Python 系统,只把经过测量确认的热点函数迁移到 Rust,再通过 PyO3 以普通 Python 模块的形式接回原有代码。

这种增量式 FFI 重构把风险限制在函数边界内。团队可以继续使用现有部署方式、测试体系和业务代码,同时获得 Rust 在 CPU 密集型任务上的性能与内存控制能力,也不必为了语言隔离立即引入微服务及其网络、部署和可观测性成本。

先找热点,再决定迁移边界

Rust 并不会自动让整个 Python 服务变快。如果请求时间主要消耗在数据库、外部 API 或磁盘 I/O 上,改写计算函数通常收效有限。迁移前应先用 cProfile、py-spy 或生产环境指标确认时间究竟花在哪里:

python -m cProfile -s cumulative app.py

# 安装后可对正在运行的进程采样
pip install py-spy
py-spy top --pid <PYTHON_PID>

适合迁移到 Rust 的代码通常具备这些特征:

  • CPU 占用高,并且会被频繁调用。
  • 输入和输出边界清晰,例如字节数组、数字列表或结构化记录。
  • 计算逻辑相对稳定,不依赖大量 Python 对象的动态行为。
  • 单次调用包含足够多的工作,可以摊薄 Python 与 Rust 之间的转换成本。

边界设计非常关键。与其从 Python 循环中调用数百万次 Rust 小函数,更适合把完整的一批数据一次性交给 Rust。FFI 调用次数、对象分配和数据复制都可能抵消底层计算带来的收益。

一个可以运行的 PyO3 最小项目

下面是一个可直接改造的示例。它把平方和计算放进 Rust,并在计算期间释放 Python GIL。示例假设本机已经安装 Python 3、Rust 和 Cargo。

项目包含三个文件:

fastmath/
├── Cargo.toml
├── pyproject.toml
└── src/
    └── lib.rs

pyproject.toml

[build-system]
requires = ["maturin>=1.5,<2.0"]
build-backend = "maturin"

[project]
name = "fastmath"
version = "0.1.0"
requires-python = ">=3.9"

[tool.maturin]
features = ["pyo3/extension-module"]

Cargo.toml

[package]
name = "fastmath"
version = "0.1.0"
edition = "2021"

[lib]
name = "fastmath"
crate-type = ["cdylib"]

[dependencies]
pyo3 = { version = "0.22", features = ["extension-module"] }

src/lib.rs

use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;

#[pyfunction]
fn sum_squares(py: Python<'_>, n: u64) -> PyResult<u64> {
    // 保证中间结果不会超出 u64;实际项目应按业务范围设计类型。
    if n > 3_000_000 {
        return Err(PyValueError::new_err("n must be <= 3,000,000"));
    }

    let total = py.allow_threads(|| {
        let mut total = 0_u64;
        for i in 0..n {
            total += i * i;
        }
        total
    });

    Ok(total)
}

#[pymodule]
fn fastmath(module: &Bound<'_, PyModule>) -> PyResult<()> {
    module.add_function(wrap_pyfunction!(sum_squares, module)?)?;
    Ok(())
}

创建虚拟环境并以 release 模式编译扩展:

cd fastmath
python -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip maturin pytest
maturin develop --release

python -c "import fastmath; print(fastmath.sum_squares(10))"

最后一条命令应输出 285。Windows PowerShell 中激活环境的命令应改为 .venv\Scripts\Activate.ps1

这里的 py.allow_threads 很重要:Rust 执行不访问 Python 对象的长时间计算时,可以释放 GIL,让其他 Python 线程继续运行。只有当闭包内的数据满足线程安全要求,并且不调用 Python API 时,才应该这样做。

用同一组测试约束两种实现

增量替换的核心不是“Rust 版本能运行”,而是新旧实现对业务输入保持相同行为。迁移期间可以保留 Python 参考实现,并用参数化测试比较结果:

# test_fastmath.py
import fastmath
import pytest


def sum_squares_python(n: int) -> int:
    if n > 3_000_000:
        raise ValueError("n must be <= 3,000,000")
    return sum(i * i for i in range(n))


@pytest.mark.parametrize("n", [0, 1, 10, 10_000, 1_000_000])
def test_rust_matches_python(n: int) -> None:
    assert fastmath.sum_squares(n) == sum_squares_python(n)


def test_invalid_range() -> None:
    with pytest.raises(ValueError, match="3,000,000"):
        fastmath.sum_squares(3_000_001)

运行测试:

pytest -q

真实项目还应覆盖空输入、最大值、Unicode、非法类型、异常映射以及并发调用。若函数处理金额、时间或二进制协议,还需要固定测试向量,避免两种语言在整数溢出、浮点舍入和编码规则上产生差异。

性能验证也必须使用 release 构建。Rust debug 构建包含额外检查,不能代表生产表现。可以这样做一个最低限度的本地比较:

# benchmark.py
from time import perf_counter
import fastmath


def python_version(n: int) -> int:
    return sum(i * i for i in range(n))


n = 2_000_000
for name, fn in [
    ("python", python_version),
    ("rust", fastmath.sum_squares),
]:
    started = perf_counter()
    result = fn(n)
    elapsed = perf_counter() - started
    print(f"{name:>6}: {elapsed:.6f}s result={result}")
python benchmark.py

这个示例可以展示函数级差异,但不能直接推导生产环境的成本收益。上线决策应观察端到端延迟、吞吐量、CPU 时间、内存峰值和扩展模块调用频率,而不只是单次微基准结果。

发布时保留回退通道

原生扩展会引入新的工程约束:需要为目标操作系统、CPU 架构和 Python ABI 构建 wheel;Rust panic 不能穿过 FFI 边界;输入转换可能复制大块内存;调试栈也会跨越两种语言。

因此,第一批迁移适合放在特性开关之后:

import os


def sum_squares(n: int) -> int:
    use_rust = os.getenv("USE_RUST_FASTMATH", "0") == "1"
    if use_rust:
        import fastmath
        return fastmath.sum_squares(n)
    return sum(i * i for i in range(n))

团队可以先在 CI 中对比两种实现,再逐步扩大生产流量。监控中应区分 Rust 路径和 Python 路径,并记录延迟、错误率、CPU 使用量和回退次数。确认收益稳定后,再删除旧实现和开关。

采用前的检查清单

  • 性能分析已经证明瓶颈位于 CPU 计算,而不是外部 I/O。
  • FFI API 使用少量、清晰、可版本化的数据类型。
  • 每次跨语言调用包含足够多的工作。
  • Python 与 Rust 实现共享一致性测试和边界测试。
  • 基准使用 release 构建,并测量端到端业务指标。
  • CI 能为所有生产平台构建并测试 wheel。
  • 异常、整数范围、内存复制和 panic 行为都有明确约定。
  • 上线过程有特性开关、指标对照和快速回退方案。

增量集成 Rust 的价值不只来自某个函数的显著加速,更来自可控的改造范围。把性能热点逐个替换、持续验证业务等价性,并用生产指标核算基础设施收益,通常比一次高风险重写更容易交付,也更容易撤回。


相关推荐