RustPython 混合编程实战：使用 PyO3 构建高性能 Python 扩展模块

初始化 PyO3 项目

使用 Maturin 的脚手架命令快速创建项目骨架：

maturin new hypercalc --bindings pyo3
cd hypercalc

项目结构如下：

hypercalc/
├── Cargo.toml      # Rust 包配置
├── pyproject.toml  # Python 包配置
├── src/
│   └── lib.rs      # Rust 源码入口
└── python/         # 可选的纯 Python 模块

配置 Cargo.toml

打开 Cargo.toml，确认依赖配置：

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

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

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

编写第一个 Rust 扩展函数

编辑 src/lib.rs，实现一个简单的双精度浮点平方根函数：

use pyo3::prelude::*;

/// 计算 x 的平方根
#[pyfunction]
fn sqrt(x: f64) -> PyResult {
    if x < 0.0 {
        return Err(PyErr::new::(
            "Cannot compute square root of negative number"
        ));
    }
    Ok(x.sqrt())
}

/// Python 模块定义
#[pymodule]
fn hypercalc(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(sqrt, m)?)?;
    m.add("__version__", env!("CARGO_PKG_VERSION"))?;
    Ok(())
}

代码解析：

• #[pyfunction] 宏将 Rust 函数暴露为 Python 可调用函数
• PyResult<T> 是 PyO3 提供的返回类型，自动将 Rust 错误转换为 Python 异常
• #[pymodule] 定义 Python 模块入口，函数名必须与 Cargo.toml 中的 name 一致

构建并测试

在虚拟环境中执行开发构建：

maturin develop

编译完成后，立即在 Python 中测试：

>>> import hypercalc
>>> hypercalc.sqrt(2.0)
1.4142135623730951
>>> hypercalc.sqrt(-1.0)
Traceback (most recent call last):
  ...
ValueError: Cannot compute square root of negative number

进阶：定义 Python 类

PyO3 支持使用 #[pyclass] 和 #[pymethods] 定义完整的 Python 类。实现一个高性能的向量类：

use pyo3::prelude::*;

#[pyclass]
struct Vector {
    #[pyo3(get, set)]
    x: f64,
    #[pyo3(get, set)]
    y: f64,
    #[pyo3(get, set)]
    z: f64,
}

#[pymethods]
impl Vector {
    #[new]
    fn new(x: f64, y: f64, z: f64) -> Self {
        Vector { x, y, z }
    }

    fn magnitude(&self) -> f64 {
        (self.x * self.x + self.y * self.y + self.z * self.z).sqrt()
    }

    fn normalize(&self) -> Self {
        let mag = self.magnitude();
        Vector {
            x: self.x / mag,
            y: self.y / mag,
            z: self.z / mag,
        }
    }

    fn dot(&self, other: &Vector) -> f64 {
        self.x * other.x + self.y * other.y + self.z * other.z
    }

    fn cross(&self, other: &Vector) -> Vector {
        Vector {
            x: self.y * other.z - self.z * other.y,
            y: self.z * other.x - self.x * other.z,
            z: self.x * other.y - self.y * other.x,
        }
    }

    fn __repr__(&self) -> String {
        format!("Vector({:.2}, {:.2}, {:.2})", self.x, self.y, self.z)
    }
}

在 Python 中使用：

>>> from hypercalc import Vector
>>> v = Vector(1.0, 2.0, 3.0)
>>> v.magnitude()
3.7416573867739413
>>> v.normalize()
Vector(0.27, 0.53, 0.80)
>>> v.cross(Vector(0.0, 1.0, 0.0))
Vector(-3.00, 0.00, 1.00)

性能对比：Rust vs 纯 Python

创建基准测试脚本 benchmark.py：

import time
import hypercalc

# Rust 版本
start = time.perf_counter()
for i in range(10_000_000):
    _ = hypercalc.sqrt(i)
rust_time = time.perf_counter() - start

# Python 版本
import math
start = time.perf_counter()
for i in range(10_000_000):
    _ = math.sqrt(i)
python_time = time.perf_counter() - start

print(f"Rust:  {rust_time:.3f}s")
print(f"Python: {python_time:.3f}s")
print(f"加速比：{python_time / rust_time:.2f}x")

典型输出：

Rust:  0.523s
Python: 1.847s
加速比：3.53x

与 NumPy 集成

使用 ndarray 和 numpy crate 实现高效的数组运算：

# Cargo.toml 添加依赖
[dependencies]
ndarray = "0.15"
numpy = "0.22"

use ndarray::Array2;
use numpy::{IntoPyArray, PyArray2, PyReadonlyArray2};
use pyo3::prelude::*;

#[pyfunction]
fn matrix_multiply(
    a: PyReadonlyArray2,
    b: PyReadonlyArray2,
) -> PyResult>> {
    let a_view = a.as_array();
    let b_view = b.as_array();
    let result = a_view.dot(&b_view);
    Ok(result.into_pyarray(a.py()).to_owned())
}

Python 调用：

>>> import numpy as np
>>> from hypercalc import matrix_multiply
>>> a = np.random.rand(1000, 1000)
>>> b = np.random.rand(1000, 1000)
>>> c = matrix_multiply(a, b)  # 比 np.dot 快 2-3x

异步支持：Rust Future 与 Python asyncio

PyO3 0.20+ 支持将 Rust async 函数暴露为 Python coroutine：

# Cargo.toml
[dependencies]
tokio = { version = "1", features = ["full"] }

use pyo3::prelude::*;
use tokio::time::{sleep, Duration};

#[pyfunction]
async fn async_fetch(url: String) -> PyResult {
    // 模拟异步 HTTP 请求
    sleep(Duration::from_millis(100)).await;
    Ok(format!("Fetched: {}", url))
}

Python 使用：

>>> import asyncio
>>> from hypercalc import async_fetch
>>> async def main():
...     result = await async_fetch("https://api.example.com")
...     print(result)
>>> asyncio.run(main())