返回文章列表

Rust FFI 实战:让 Rust 和 Python/Node.js 组队干活

1224·9 分钟阅读
RustFFIPythonNode.js跨语言调用

ruff 比 flake8 快 10-100 倍,polars 比 pandas 快 5-50 倍,orjson 比标准库 json 快 10 倍。你有没有想过,这些 Python 库凭什么这么快?

答案不是什么黑魔法,而是它们的性能核心都是 Rust

但有意思的是,这些库的用户完全感知不到 Rust 的存在——import ruffimport polars,用法和纯 Python 库一模一样。这就是 Rust FFI(Foreign Function Interface)的魅力:你不需要用 Rust 重写整个项目,只需要把最慢的那部分用 Rust 实现,然后像调普通函数一样调用它。

我之前有个 Python 脚本,批量处理大量文本文件做关键词统计。10 万个文件跑一次要 40 秒。同事说你换成 Rust 重写吧,我说不——我只改了几十行核心代码,让 Rust 来干脏活,Python 依然负责胶水逻辑。最终同样的任务,快了几十倍。

这篇文章就来手把手教你:怎么用 PyO3 给 Python 加速,怎么用 napi-rs 给 Node.js 加速,以及 FFI 这条路上有哪些坑等着你。

先搞明白:FFI 到底是什么

FFI = Foreign Function Interface,翻译过来就是"外部函数接口"。说白了就是让一种语言写的代码能调用另一种语言写的函数

我们这里关注的场景是:

Python/Node.js(上层,胶水逻辑)
    ↓ 调用
Rust(底层,性能核心)

这是最常见的方向——上层语言负责业务逻辑和生态,Rust 负责把计算密集的部分干到极致。反过来(Rust 调用 Python/Node)也有,但用得少很多。

生态全景

目前主流的 Rust FFI 方案:

方案 Stars 目标语言 上手难度 适用场景
PyO3 + maturin 15.8k / 5.6k Python ⭐⭐ 数据处理、ML 工具链
napi-rs 7.8k Node.js ⭐⭐ 前端工具链、构建工具
neon 8.4k Node.js ⭐⭐⭐ Node.js 原生模块
uniffi 4.7k Python/JS/Kotlin/Swift ⭐⭐ 多平台 SDK

(Stars 数据截至 2026 年 6 月)

选哪个很简单:Python 生态用 PyO3,Node.js 生态用 napi-rs。neon 也可以做 Node.js 绑定,但 napi-rs 的开发体验更好,生态也在快速增长——rspack、rolldown、swc 这些前端工具链的明星项目都选了它。uniffi 是 Mozilla 出品,适合需要同时支持 Python/JS/Kotlin/Swift 多语言的场景,比如你做一个跨平台 SDK。

哪些知名项目在用

PyO3 生态(Python 侧):

  • polars — DataFrame 库,号称比 pandas 快 10-50 倍
  • ruff — Python linter,比 flake8 快 10-100 倍
  • pydantic-core — Pydantic V2 的验证核心,比 V1 快 5-50 倍
  • orjson — JSON 库,比标准库 json 快 10 倍
  • tiktoken — OpenAI 的 tokenizer,纯 Rust 实现

napi-rs 生态(Node.js 侧):

  • rspack — Webpack 替代品,Rust 实现的打包器
  • rolldown — Rollup 的 Rust 替代品
  • swc — Babel/Terser 的 Rust 替代品
  • oxlint — ESLint 的 Rust 替代品

这些项目证明了一件事:Rust FFI 不是玩具,是经过大规模生产验证的方案。

PyO3 实战:给 Python 装个涡轮增压

废话不多说,直接上手。我们做一个真实的场景:批量读取大量文本文件,用分词提取关键词,统计词频

环境搭建

mkdir text-boost && cd text-boost
python -m venv .env
source .env/bin/activate
pip install maturin
maturin init --bindings pyo3

maturin init 会生成一个最小的 PyO3 项目,看看它长什么样:

text-boost/
├── Cargo.toml          # Rust 依赖配置
├── pyproject.toml       # Python 包配置
└── src/
    └── lib.rs           # Rust 代码写这里

Cargo.toml 里已经有 PyO3 依赖了:

[package]
name = "text-boost"
version = "0.1.0"
edition = "2024"
 
[lib]
name = "text_boost"
crate-type = ["cdylib"]   # 编译成动态库,Python 才能 import
 
[dependencies]
pyo3 = "0.28"

crate-type = ["cdylib"] 这一行很关键——它告诉 Rust 编译器输出一个动态库(.so / .pyd),而不是普通的 Rust 库。没有它,Python 没法 import。

第一版:纯 Python 基准

先写一个纯 Python 版本,作为性能基准:

# benchmark_pure.py
import os
import time
from collections import Counter
 
def extract_keywords(text: str) -> list[str]:
    """简单的中英文关键词提取(按空格和标点切分)"""
    import re
    # 粗暴切分:按非字母中文字符切
    tokens = re.findall(r'[一-鿿]+|[a-zA-Z]+', text)
    # 过滤太短的词
    return [t for t in tokens if len(t) >= 2]
 
def process_files(root_dir: str) -> dict[str, int]:
    counter = Counter()
    for dirpath, _, filenames in os.walk(root_dir):
        for fname in filenames:
            if not fname.endswith(('.txt', '.md')):
                continue
            fpath = os.path.join(dirpath, fname)
            try:
                with open(fpath, 'r', encoding='utf-8') as f:
                    text = f.read()
                keywords = extract_keywords(text)
                counter.update(keywords)
            except (UnicodeDecodeError, PermissionError):
                pass
    return dict(counter)
 
if __name__ == '__main__':
    start = time.time()
    result = process_files('./test_data')
    elapsed = time.time() - start
    top10 = sorted(result.items(), key=lambda x: -x[1])[:10]
    print(f"耗时: {elapsed:.2f}s")
    print(f"关键词总数: {len(result)}")
    print(f"Top 10: {top10}")

能跑,但慢。瓶颈在哪?文本读取 + 正则切分 + 统计,全是 CPU 密集操作。这正是 Rust 的主场。

第二版:用 PyO3 写 Rust 加速

现在把核心逻辑搬到 Rust 里。编辑 src/lib.rs

use pyo3::prelude::*;
use std::collections::HashMap;
use std::fs;
use std::path::Path;
use walkdir::WalkDir;
 
/// 从文本中提取关键词(简单的中英文切分)
fn extract_keywords(text: &str) -> Vec<String> {
    let mut keywords = Vec::new();
    let mut current = String::new();
    let mut is_chinese = false;
 
    for ch in text.chars() {
        if ch.is_ascii_alphabetic() {
            if is_chinese && current.len() >= 2 {
                keywords.push(std::mem::take(&mut current));
            }
            is_chinese = false;
            current.push(ch);
        } else if ch >= '\u{4e00}' && ch <= '\u{9fff}' {
            if !is_chinese && current.len() >= 2 {
                keywords.push(std::mem::take(&mut current));
            }
            is_chinese = true;
            current.push(ch);
        } else {
            if current.len() >= 2 {
                keywords.push(std::mem::take(&mut current));
            }
        }
    }
    if current.len() >= 2 {
        keywords.push(current);
    }
    keywords
}
 
/// 批量处理文本文件,统计词频
#[pyfunction]
fn process_files(root_dir: &str) -> PyResult<HashMap<String, usize>> {
    let mut counter: HashMap<String, usize> = HashMap::new();
 
    for entry in WalkDir::new(root_dir)
        .into_iter()
        .filter_map(|e| e.ok())
    {
        let path = entry.path();
        // 只处理文本文件
        let is_text = path.extension().map_or(false, |ext| {
            ext == "txt" || ext == "md"
        });
        if !is_text {
            continue;
        }
 
        if let Ok(content) = fs::read_to_string(path) {
            for keyword in extract_keywords(&content) {
                *counter.entry(keyword).or_insert(0) += 1;
            }
        }
    }
 
    Ok(counter)
}
 
/// Rust 模块定义——Python import 的入口
#[pymodule]
fn text_boost(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(process_files, m)?)?;
    Ok(())
}

然后在 Cargo.toml 里加上需要的依赖:

[dependencies]
pyo3 = { version = "0.28", features = ["extension-module"] }
walkdir = "2"

编译并安装到当前 Python 环境:

maturin develop --release

--release 很重要! 不加的话 Rust 会用 debug 模式编译,性能可能还不如 Python。我第一次跑的时候忘了加 --release,结果 Rust 版本比 Python 还慢,差点怀疑人生。

现在可以在 Python 里直接调用了:

# benchmark_rust.py
import time
from text_boost import process_files   # 直接 import,和普通 Python 模块一样
 
start = time.time()
result = process_files('./test_data')
elapsed = time.time() - start
top10 = sorted(result.items(), key=lambda x: -x[1])[:10]
print(f"耗时: {elapsed:.2f}s")
print(f"关键词总数: {len(result)}")
print(f"Top 10: {top10}")

如何自己跑 benchmark

别信别人嘴里的数字,自己跑才准。这里介绍两种方式:

方式一:Python 内置 time

import time
 
# 跑 3 次取平均
times = []
for _ in range(3):
    start = time.time()
    process_files('./test_data')
    times.append(time.time() - start)
 
print(f"平均耗时: {sum(times)/len(times):.2f}s")

方式二:hyperfine(推荐)

# 安装 hyperfine
brew install hyperfine   # macOS
# 或 cargo install hyperfine
 
# 对比两个版本,各跑 5 次,预热 2 次
hyperfine \
  'python benchmark_pure.py' \
  'python benchmark_rust.py' \
  --warmup 2 \
  --runs 5

hyperfine 会给出统计分布(中位数、标准差),比单次计时靠谱得多。

真实世界的加速倍数取决于你的场景。 文本处理这种纯 CPU 任务,通常能看到 10-50 倍的提升。如果是 I/O 密集(比如大量网络请求),加速比会小很多,因为瓶颈不在计算上。这正是 ruff、polars、orjson 这些项目选择 Rust 的原因——它们的瓶颈恰好在 CPU 上。

进阶:暴露 Python 类

上面的例子里,我们只暴露了一个函数。如果你想暴露一个有状态的对象(比如带配置的搜索引擎),可以用 #[pyclass]

use pyo3::prelude::*;
use std::collections::HashMap;
 
#[pyclass]
struct TextProcessor {
    min_word_length: usize,
    case_sensitive: bool,
    counter: HashMap<String, usize>,
}
 
#[pymethods]
impl TextProcessor {
    #[new]
    fn new(min_word_length: Option<usize>, case_sensitive: Option<bool>) -> Self {
        Self {
            min_word_length: min_word_length.unwrap_or(2),
            case_sensitive: case_sensitive.unwrap_or(false),
            counter: HashMap::new(),
        }
    }
 
    /// 处理单个文本
    fn process_text(&mut self, text: &str) {
        for keyword in self.extract(text) {
            *self.counter.entry(keyword).or_insert(0) += 1;
        }
    }
 
    /// 获取词频最高的 N 个关键词
    fn top_keywords(&self, n: usize) -> Vec<(String, usize)> {
        let mut items: Vec<_> = self.counter.iter().collect();
        items.sort_by(|a, b| b.1.cmp(a.1));
        items.into_iter()
            .take(n)
            .map(|(k, v)| (k.clone(), *v))
            .collect()
    }
 
    /// 获取关键词总数
    #[getter]
    fn keyword_count(&self) -> usize {
        self.counter.len()
    }
}
 
impl TextProcessor {
    fn extract(&self, text: &str) -> Vec<String> {
        // 和之前一样的切分逻辑,但用 self.min_word_length 过滤
        let mut keywords = Vec::new();
        let mut current = String::new();
        let mut is_chinese = false;
 
        for ch in text.chars() {
            let ch = if self.case_sensitive { ch } else { ch.to_ascii_lowercase() };
            if ch.is_ascii_alphabetic() {
                if is_chinese && current.len() >= self.min_word_length {
                    keywords.push(std::mem::take(&mut current));
                }
                is_chinese = false;
                current.push(ch);
            } else if ch >= '\u{4e00}' && ch <= '\u{9fff}' {
                if !is_chinese && current.len() >= self.min_word_length {
                    keywords.push(std::mem::take(&mut current));
                }
                is_chinese = true;
                current.push(ch);
            } else {
                if current.len() >= self.min_word_length {
                    keywords.push(std::mem::take(&mut current));
                }
            }
        }
        if current.len() >= self.min_word_length {
            keywords.push(current);
        }
        keywords
    }
}

Python 侧用起来是这样的:

from text_boost import TextProcessor
 
proc = TextProcessor(min_word_length=3, case_sensitive=False)
proc.process_text("用Rust实现高性能优化方案")
proc.process_text("Rust的内存安全机制是核心优势")
 
print(proc.top_keywords(5))
# [('Rust', 2), ('高性能', 1), ('实现', 1), ('优化', 1), ('方案', 1)]
print(proc.keyword_count)
# 8

#[new] 对应 Python 的 __init__#[getter] 把方法变成属性。PyO3 的宏设计得很 Pythonic,写起来很自然。

napi-rs 实战:给 Node.js 也来一个

Node.js 生态也有同样的需求。napi-rs 的体验和 PyO3 很像,但 API 风格有些不同。

环境搭建

# 需要 Node.js 16+
npx @napi-rs/cli init text-boost-node
cd text-boost-node
npm install

生成的项目结构:

text-boost-node/
├── Cargo.toml
├── package.json
├── build.rs              # napi 的构建脚本
├── index.d.ts            # 自动生成的 TypeScript 类型
├── index.js              # 自动生成的 JS 入口
└── src/
    └── lib.rs             # Rust 代码写这里

Rust 侧代码

src/lib.rs

use napi_derive::napi;
use std::collections::HashMap;
 
fn extract_keywords(text: &str, min_len: usize) -> Vec<String> {
    let mut keywords = Vec::new();
    let mut current = String::new();
    let mut is_chinese = false;
 
    for ch in text.chars() {
        if ch.is_ascii_alphabetic() {
            if is_chinese && current.len() >= min_len {
                keywords.push(std::mem::take(&mut current));
            }
            is_chinese = false;
            current.push(ch);
        } else if ch >= '\u{4e00}' && ch <= '\u{9fff}' {
            if !is_chinese && current.len() >= min_len {
                keywords.push(std::mem::take(&mut current));
            }
            is_chinese = true;
            current.push(ch);
        } else {
            if current.len() >= min_len {
                keywords.push(std::mem::take(&mut current));
            }
        }
    }
    if current.len() >= min_len {
        keywords.push(current);
    }
    keywords
}
 
/// 批量处理文本,统计词频
#[napi]
fn process_texts(texts: Vec<String>, min_word_length: Option<u32>) -> HashMap<String, u32> {
    let min_len = min_word_length.unwrap_or(2) as usize;
    let mut counter: HashMap<String, u32> = HashMap::new();
 
    for text in texts {
        for keyword in extract_keywords(&text, min_len) {
            *counter.entry(keyword).or_insert(0) += 1;
        }
    }
 
    counter
}

Node.js 侧调用

// benchmark.mjs
import { processTexts } from './index.js';
 
const texts = [
    '用Rust实现高性能优化方案',
    'Rust的内存安全机制是核心优势',
    '高性能计算是未来趋势',
];
 
const result = processTexts(texts, 2);
console.log(result);
// { 'Rust': 3, '高性能': 2, '实现': 1, '优化': 1, '方案': 1, ... }

编译运行:

npm run build       # 编译 Rust 代码
node benchmark.mjs  # 运行

PyO3 vs napi-rs:API 风格对比

两个框架的思路很像,但写法有差异:

维度 PyO3 napi-rs
函数标记 #[pyfunction] #[napi]
类标记 #[pyclass] + #[pymethods] #[napi] 直接在 impl 上
构造函数 #[new] #[napi(constructor)]
生命周期 需要 Python<'py> 参数 不需要,更简洁
错误处理 PyResult<T> → Python Exception Result<T> → JS Error
GIL 管理 需要 Python::with_gil 不需要(Node.js 没有 GIL)

最大的区别在 GIL 上。Python 有一个全局解释器锁(GIL),PyO3 里到处都要和它打交道。Node.js 没有这个问题,所以 napi-rs 的代码写起来更"普通 Rust"一些。

踩坑实录:FFI 不是银弹

讲完怎么用,该讲讲路上有哪些坑了。这些都是我实际踩过的,希望你不用再踩一遍。

坑一:GIL 是 Python FFI 的紧箍咒

Python 的 GIL(Global Interpreter Lock)意味着同一时刻只有一个线程能执行 Python 字节节码。PyO3 的所有操作都需要持有 GIL。

如果你在 Rust 里用 rayon 并行处理数据,处理完之后要把结果返回给 Python,你需要这样写:

use pyo3::prelude::*;
use rayon::prelude::*;
 
#[pyfunction]
fn parallel_process(py: Python<'_>, texts: Vec<String>) -> PyResult<HashMap<String, usize>> {
    // 先释放 GIL,让 Rust 自由并行
    let result = py.allow_threads(|| {
        // 这段代码里没有 Python 对象,可以自由并行
        let mut counter: HashMap<String, usize> = HashMap::new();
        let results: Vec<_> = texts.par_iter()
            .map(|text| extract_keywords(text))
            .collect();
 
        for keywords in results {
            for kw in keywords {
                *counter.entry(kw).or_insert(0) += 1;
            }
        }
        counter
    });
 
    Ok(result)
}

py.allow_threads(|| { ... }) 告诉 Python:"我这块代码不用 Python 对象,你把 GIL 放开,让别的线程也能跑。" 如果不放 GIL,你的 par_iter 并行了个寂寞——所有线程排队等 GIL。

经验法则: Rust 里做计算的时候释放 GIL,需要返回 Python 对象的时候再拿回 GIL。PyO3 会在编译期帮你检查大部分 GIL 相关的错误,但有些只能在运行时发现。

坑二:类型转换的隐性开销

Python 的 list → Rust 的 Vec,Python 的 dict → Rust 的 HashMap——这些转换不是免费的。如果传一个包含 100 万个元素的 list,每个元素都要从 Python 对象转成 Rust 对象,这个开销可能很大。

零拷贝的技巧:

use pyo3::types::PyBytes;
 
#[pyfunction]
fn process_bytes(py: Python<'_>, data: &Bound<'_, PyBytes>) -> PyResult<usize> {
    // 直接拿到 bytes 的引用,不拷贝
    let bytes = data.as_bytes();
    Ok(bytes.len())
}

对于大数据,尽量传 bytes / Buffer 而不是 str / string,避免编码转换开销。

坑三:Rust panic 会变成 segfault

如果 Rust 代码 panic 了,Python 进程会直接 segfault 崩掉——没有友好的异常信息,没有 traceback,就是一坨 core dump。

解决方案:用 catch_unwind 兜底

use std::panic;
 
#[pyfunction]
fn safe_process(text: &str) -> PyResult<String> {
    let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
        // 你的可能 panic 的代码
        extract_keywords(text)
    }));
 
    match result {
        Ok(keywords) => Ok(keywords.join(", ")),
        Err(e) => Err(pyo3::exceptions::PyRuntimeError::new_err(
            format!("Rust panic: {:?}", e)
        ))
    }
}

这样 Rust panic 会变成 Python 的 RuntimeError,至少有错误信息可以看。

更好的做法: 在 Rust 代码里尽量用 Result 而不是 unwrap()unwrap() 是定时炸弹,FFI 边界上尤其危险。

坑四:发布和分发

本地开发用 maturin develop 很爽,但要发布到 PyPI / npm 让别人用,就得处理跨平台问题。

Python 侧(maturin):

# 本地构建当前平台的 wheel
maturin build --release
 
# 跨平台构建(推荐用 GitHub Actions)
# .github/workflows/release.yml 里用 maturin-action

maturin-action 会自动在 Linux(manylinux)、macOS、Windows 上构建 wheel,发布到 PyPI。

Node.js 侧(napi-rs):

napi-rs 更贴心——它会为每个平台生成 prebuild 的二进制文件,用户 npm install 的时候自动下载对应平台的版本,不需要本地编译。

# 生成各平台的 prebuild
npx napi build --platform
 
# 发布到 npm
npm publish

常见坑:

  • Linux 上要遵守 manylinux 规范,否则用户装不了。maturin 有内置的 auditwheel 检查
  • macOS 要同时支持 x86_64 和 aarch64(Intel 和 Apple Silicon)
  • Windows 需要 MSVC 工具链
  • 建议从一开始就用 CI 做跨平台构建,别想着"我先在本机调通再说"

什么时候该用 FFI,什么时候别折腾

FFI 很酷,但不是万能药。用对了事半功倍,用错了徒增复杂度。

✅ 该用的场景

  • CPU 密集型计算 — 数据处理、编解码、加密、压缩、搜索索引。瓶颈在算力上,Rust 的零成本抽象能带来实打实的提升
  • 需要 Rust 生态的库 — 比如你想用 serde 处理 JSON、用 reqwest 做 HTTP、用 tantivy 做全文搜索,但你的主项目是 Python/Node.js
  • 性能瓶颈明确 — 你 profile 过了,知道 80% 的时间花在某几个函数上,只加速这几个函数就够了

❌ 不该用的场景

  • I/O 密集型 — 大量网络请求、数据库查询。瓶颈不在语言上,换 Rust 也不会快多少,反而增加了维护成本
  • 团队没人懂 Rust — FFI 代码需要同时懂两种语言和它们的交互方式。如果团队里没人能维护 Rust 代码,这就是技术债
  • 原型阶段 — 先让它跑起来,再让它跑得快。过早优化是万恶之源

渐进式迁移策略

FFI 的精髓不是"重写",而是"嫁接":

第一步:Profile,找到真正的瓶颈函数
第二步:只把瓶颈函数用 Rust 实现
第三步:通过 FFI 在原项目中调用
第四步:验证性能提升,确认功能一致
第五步:逐步扩大 Rust 的范围(如果需要)

这就是 polars、ruff、pydantic-core 走过的路——它们不是一夜之间用 Rust 重写的,而是先从最核心的性能热点开始,逐步把 Rust 的领地扩大。

还有第三条路:WASM

如果你的场景是前端(浏览器),还有一个选择:Rust → WebAssembly。我在之前的文章里写过 Rust WASM 的实战(参见《Rust WASM 实战:我用它给前端加了个涡轮增压》),它和 FFI 的区别是:

维度 FFI(PyO3 / napi-rs) WASM
运行环境 原生进程 浏览器 / WASI 运行时
性能 接近原生 接近原生(有少量沙箱开销)
部署方式 需要编译对应平台的二进制 一个 .wasm 文件走天下
适用场景 服务端、CLI 工具、桌面应用 前端、插件系统、跨平台

简单说:服务端用 FFI,前端用 WASM,两个不冲突。

总结

回到开头的问题:ruff、polars、orjson 凭什么那么快?因为它们找到了性能瓶颈,然后用 Rust 精准打击。

你也可以这么做:

  1. Profile 你的代码,找到真正慢的地方
  2. 用 PyO3 或 napi-rs 把那部分用 Rust 重写
  3. maturin developnpm run build 编译,Python/Node 侧无感调用
  4. 跑 benchmark,确认提升

FFI 不是重写,是嫁接——把 Rust 的性能根系接到 Python/Node.js 的枝干上。果实是你的,树还是原来的树。