Python 生态扩展 / C 扩展
Python 作为胶水语言,既可以作为胶水粘合现有 C/C++ 库,也可以作为宿主语言通过 C 扩展机制获得原生性能。本文档系统介绍 Python 包开发、C 扩展开发(Python C API)、Cython、CFFI 以及 PyPy JIT 编译五个方面的技术要点。
1. Python 包开发
1.1 项目结构
一个标准的 Python 包项目结构如下:
my-package/
├── pyproject.toml # 项目构建配置(PEP 621)
├── setup.cfg # 传统配置(setuptools 可选)
├── setup.py # 传统构建脚本(可选)
├── README.md
├── LICENSE
├── src/ # 源码目录(推荐 src layout)
│ └── my_package/ # Python 包目录
│ ├── __init__.py # 包初始化,导出公共 API
│ ├── core.py
│ ├── utils.py
│ └── sub_package/ # 子包
│ ├── __init__.py
│ └── ...
├── tests/ # 测试目录
│ ├── __init__.py
│ ├── test_core.py
│ └── test_utils.py
└── docs/ # 文档目录1.1.1 src 目录布局(src layout)
src layout 是 PEP 517/518 推荐的源码组织方式。将实际包代码放在 src/ 目录下,可以避免在项目根目录直接导入包时产生的隐式导入问题。
project-root/
├── src/
│ └── my_package/
│ ├── __init__.py
│ └── module.py
├── tests/
└── pyproject.toml1.1.2 __init__.py 包组织
__init__.py 控制包级别的导入行为:
# src/my_package/__init__.py
from .core import main_function
from .utils import helper
__version__ = "1.0.0"
__all__ = ["main_function", "helper"]1.2 pyproject.toml 完整配置
pyproject.toml 是 PEP 621 定义的现代 Python 项目配置标准,涵盖构建系统、项目元数据、依赖、入口点等全部信息。
[build-system]
requires = ["setuptools>=68.0", "wheel>=0.42"]
build-backend = "setuptools.build_meta"
[project]
name = "my-package"
version = "1.0.0"
description = "A sample Python package"
readme = "README.md"
requires-python = ">=3.10"
license = {text = "MIT"}
authors = [
{name = "Author Name", email = "author@example.com"},
]
maintainers = [
{name = "Maintainer Name", email = "maintainer@example.com"},
]
keywords = ["python", "package", "example"]
classifiers = [
"Development Status :: 5 - Production/Stable",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries :: Python Modules",
]
[dependencies]
numpy = ">=1.24"
requests = ">=2.31"
pydantic = ">=2.0"
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-cov>=4.0",
"black>=24.0",
"ruff>=0.1",
"mypy>=1.0",
]
docs = [
"sphinx>=7.0",
"sphinx-rtd-theme>=2.0",
]
test = [
"pytest>=8.0",
"pytest-cov>=4.0",
]
[project.entry-points]
# 控制台脚本(CLI 入口)
console_scripts = [
"my-cli = my_package.cli:main",
]
[project.scripts]
# 另一种脚本定义方式
my-script = "my_package.script:run"
[project.gui-scripts]
# GUI 应用程序入口(Windows 平台)
my-gui = "my_package.gui:launch"
[tool.setuptools]
package-dir = {"" = "src"}
packages = {find = {where = ["src"]}}
[tool.setuptools.package-data]
"*" = ["*.txt", "*.json"]
"my_package" = ["data/**/*"]
[tool.pytest.ini_options]
testpaths = ["tests"]
[tool.black]
target-version = ["py310"]
line-length = 100
[tool.ruff]
target-version = "py310"
line-length = 1001.3 构建工具选型
Python 生态中主流的构建工具包括 setuptools、Poetry、Flit 和 PDM。以下是详细对比:
| 特性 | setuptools | Poetry | Flit | PDM |
|---|---|---|---|---|
| 构建后端 | setuptools.build_meta | poetry.core.masonry.api | flit_core.buildapi | pdm.pep517.api |
| 配置文件 | setup.cfg/setup.py/setuptools config | pyproject.toml (tool.poetry) | pyproject.toml (tool.flit) | pyproject.toml (tool.pdm) |
| 依赖解析 | 无内置解析器 | 有(PEP 508 格式) | 无 | 有(PEP 508 + lock) |
| 锁文件 | 不支持 | poetry.lock | 不支持 | pdm.lock |
| 虚拟环境管理 | 不支持 | poetry shell | 不支持 | pdm venv |
| 发布流程 | build + twine | poetry publish | flit publish | pdm publish |
| 学习曲线 | 中等 | 低 | 低 | 中 |
| 社区生态 | 最成熟,生态最广 | 快速发展 | 轻量级 | 后起之秀 |
1.3.1 配置差异对比
setuptools(通过 setup.cfg):
[metadata]
name = my-package
version = 1.0.0
description = A sample Python package
[options]
packages = find:
package_dir =
= src
install_requires =
numpy>=1.24
requests>=2.31
[options.packages.find]
where = src
[options.extras_require]
dev =
pytest>=8.0
black>=24.0Poetry(通过 pyproject.toml):
[tool.poetry]
name = "my-package"
version = "1.0.0"
description = "A sample Python package"
authors = ["Author <author@example.com>"]
[tool.poetry.dependencies]
python = "^3.10"
numpy = "^1.24"
requests = "^2.31"
[tool.poetry.dev-dependencies]
pytest = "^8.0"
black = "^24.0"
[build-system]
requires = ["poetry-core>=1.0"]
build-backend = "poetry.core.masonry.api"Flit(通过 pyproject.toml):
[project]
name = "my-package"
version = "1.0.0"
description = "A sample Python package"
dependencies = [
"numpy>=1.24",
"requests>=2.31",
]
[build-system]
requires = ["flit_core>=3.8"]
build-backend = "flit_core.buildapi"
[tool.flit.module]
name = "my_package"PDM(通过 pyproject.toml):
[project]
name = "my-package"
version = "1.0.0"
description = "A sample Python package"
dependencies = [
"numpy>=1.24",
"requests>=2.31",
]
[build-system]
requires = ["pdm-pep517>=1.0"]
build-backend = "pdm.pep517.api"
[tool.pdm]
distribution = true1.3.2 锁文件与发布流程差异
| 方面 | setuptools | Poetry | Flit | PDM |
|---|---|---|---|---|
| 锁文件格式 | 无 | poetry.lock(TOML) | 无 | pdm.lock(TOML) |
| 确定性安装 | pip freeze > requirements.txt | poetry.lock 保证 | 无 | pdm.lock 保证 |
| 构建命令 | python -m build | poetry build | flit build | pdm build |
| 发布命令 | twine upload dist/* | poetry publish | flit publish | pdm publish |
| 版本锁定 | requirements.txt 手动管理 | 自动 | 无 | 自动 |
1.4 发布到 PyPI
1.4.1 Twine 发布
Twine 是发布 Python 包到 PyPI 的标准工具,支持 HTTPS 安全上传。
# 安装构建工具和 Twine
pip install build twine
# 构建分发包
python -m build
# 上传到测试 PyPI
twine upload --repository testpypi dist/*
# 上传到正式 PyPI
twine upload dist/*1.4.2 API Token 配置
推荐使用 API Token 代替用户名密码认证。在 $HOME/.pypirc 中配置:
[distutils]
index-servers =
pypi
testpypi
[pypi]
username = __token__
password = pypi-xxxxxxxxxxxxxxxxxxxx
[testpypi]
repository = https://test.pypi.org/legacy/
username = __token__
password = pypi-xxxxxxxxxxxxxxxxxxxx或通过环境变量设置:
# Windows 命令提示符
set TWINE_USERNAME=__token__
set TWINE_PASSWORD=pypi-xxxxxxxxxxxxxxxxxxxx
# 或使用 keyring(推荐,更安全)
pip install keyring
keyring set https://upload.pypi.org/legacy/ __token__1.4.3 发布完整流程
# 1. 清理旧的构建产物
rm -rf dist/ build/ *.egg-info
# 2. 安装构建依赖
pip install build twine
# 3. 构建源码分发包(sdist)和 wheel 包
python -m build
# 4. 验证包内容
twine check dist/*
# 5. 上传到测试 PyPI 验证
twine upload --repository testpypi dist/*
# 6. 安装测试
pip install --index-url https://test.pypi.org/simple/ my-package
# 7. 确认无误后上传到正式 PyPI
twine upload dist/*1.4.4 测试 PyPI
测试 PyPI 环境用于在正式发布前验证包的构建和安装过程。
# 从测试 PyPI 安装
pip install --index-url https://test.pypi.org/simple/ my-package
# 从测试 PyPI 安装(同时允许从正式 PyPI 安装依赖)
pip install --index-url https://test.pypi.org/simple/ \
--extra-index-url https://pypi.org/simple/ my-package1.4.5 版本号规范
Python 包遵循 PEP 440 版本号规范:
| 版本段 | 格式 | 示例 |
|---|---|---|
| 正式版 | N.N.N | 1.0.0, 2.3.1 |
| Alpha | N.N.NaN | 1.0.0a1 |
| Beta | N.N.NbN | 1.0.0b2 |
| 候选版 | N.N.NrcN | 1.0.0rc1 |
| 开发版 | N.N.N.devN | 1.0.0.dev2 |
| 后置版本 | N.N.N.postN | 1.0.0.post1 |
版本号递增建议:
- 修复 Bug:递增补丁号,如 1.0.0 -> 1.0.1
- 添加向下兼容的功能:递增次版本号,如 1.0.0 -> 1.1.0
- 不兼容的 API 变更:递增主版本号,如 1.0.0 -> 2.0.0
2. C 扩展开发(Python C API)
Python C API 是 CPython 最底层的扩展机制,允许编写纯 C 代码作为 Python 模块。
2.1 C 扩展结构
一个最小的 C 扩展模块包含以下要素:
// my_extension.c
#include <Python.h> // 必须最先包含
// 模块方法实现
static PyObject* my_extension_hello(PyObject* self, PyObject* args) {
// 此函数接受 Python 参数并返回 Python 对象
if (!PyArg_ParseTuple(args, "")) {
return NULL;
}
return PyUnicode_FromString("Hello from C extension!");
}
// 模块方法表
static PyMethodDef MyExtensionMethods[] = {
{"hello", my_extension_hello, METH_VARARGS, "Return a greeting message."},
{NULL, NULL, 0, NULL} // 哨兵条目,标记方法表结束
};
// 模块定义结构
static struct PyModuleDef my_extension_module = {
PyModuleDef_HEAD_INIT,
"my_extension", // 模块名
NULL, // 文档字符串
-1, // 模块状态大小(-1 表示全局状态)
MyExtensionMethods // 方法表
};
// 模块初始化函数
PyMODINIT_FUNC PyInit_my_extension(void) {
return PyModule_Create(&my_extension_module);
}2.1.1 编译配置(setuptools)
# setup.py
from setuptools import setup, Extension
my_extension = Extension(
"my_extension",
sources=["my_extension.c"],
include_dirs=[], # 额外的头文件路径
libraries=[], # 链接的外部库
library_dirs=[], # 库文件搜索路径
)
setup(
name="my-extension",
version="1.0.0",
ext_modules=[my_extension],
)2.1.2 编译配置(pyproject.toml + setup.cfg)
# pyproject.toml
[build-system]
requires = ["setuptools>=68.0", "wheel>=0.42"]
build-backend = "setuptools.build_meta"# setup.cfg
[metadata]
name = my-extension
version = 1.0.0
[options]
ext_modules = my_extension
# 通过 setup.py 定义 Extension# setup.py(用于定义 Extension,同时被 pyproject.toml 使用)
from setuptools import setup, Extension
setup(
ext_modules=[
Extension("my_extension", sources=["my_extension.c"]),
]
)2.1.3 PyMethodDef 详解
struct PyMethodDef {
const char *ml_name; // Python 中可用的方法名
PyCFunction ml_meth; // C 函数指针
int ml_flags; // 调用约定标志
const char *ml_doc; // 文档字符串
};ml_flags 常用值:
| 标志 | 说明 |
|---|---|
| METH_VARARGS | 接受位置参数,函数签名 (PyObject* self, PyObject* args) |
| METH_KEYWORDS | 接受关键字参数,函数签名 (PyObject* self, PyObject* args, PyObject* kwds) |
| METH_NOARGS | 不接受参数(args 为 NULL) |
| METH_O | 接受单个对象参数 |
| METH_VARARGS | METH_KEYWORDS | 同时支持位置和关键字参数 |
2.2 类型转换
2.2.1 PyArg_ParseTuple(Python -> C)
static PyObject* parse_demo(PyObject* self, PyObject* args) {
int int_val;
double float_val;
const char* str_val;
PyObject* obj_val;
// 解析多个参数
if (!PyArg_ParseTuple(args, "idsO",
&int_val, // i: int
&float_val, // d: double
&str_val, // s: string (const char*)
&obj_val)) { // O: PyObject*
return NULL;
}
printf("int: %d, float: %f, string: %s\n", int_val, float_val, str_val);
Py_RETURN_NONE;
}格式化字符表:
| 字符 | C 类型 | Python 类型 |
|---|---|---|
i | int | int |
l | long | int |
d | double | float |
f | float | float |
s | const char* | str(自动编码) |
s# | const char*, int | str/bytes + 长度 |
u | const Py_UNICODE* | str(Unicode) |
O | PyObject* | 任意对象 |
O& | converter, void* | 通过转换函数转换 |
z | const char* | str 或 None(None 时为 NULL) |
b | unsigned char | bytes(长度1) |
B | unsigned char | bytes(不检查范围) |
c | char | bytes(长度1) |
2.2.2 Py_BuildValue(C -> Python)
static PyObject* build_demo(PyObject* self, PyObject* args) {
int val = 42;
double pi = 3.14159;
const char* msg = "hello";
// 构建元组
return Py_BuildValue("(ids)", val, pi, msg);
// 返回: (42, 3.14159, "hello")
// 构建字典
// return Py_BuildValue("{s:i, s:d}", "count", val, "pi", pi);
// 返回: {"count": 42, "pi": 3.14159}
// 构建列表
// return Py_BuildValue("[i,i,i]", 1, 2, 3);
// 返回: [1, 2, 3]
}格式字符与 PyArg_ParseTuple 类似,但 Py_BuildValue 额外支持:
| 字符 | 说明 |
|---|---|
() | 生成元组 |
[] | 生成列表 |
{} | 生成字典 |
s# | (const char*, int) 构建字符串 |
N | 直接使用 PyObject*(偷取引用) |
O | 直接使用 PyObject*(增加引用) |
2.3 引用计数管理
Python 使用引用计数进行内存管理。C 扩展开发中需要手动管理引用计数。
2.3.1 核心宏
#include <Python.h>
void reference_demo(void) {
PyObject* obj = PyLong_FromLong(42);
// 增加引用计数
Py_INCREF(obj); // obj 引用计数 +1
Py_XINCREF(obj); // obj 可能为 NULL,引用计数 +1
// 减少引用计数(引用计数为 0 时释放对象)
Py_DECREF(obj); // obj 引用计数 -1
Py_XDECREF(obj); // obj 可能为 NULL,引用计数 -1
// 设置对象(增加新引用,减少旧引用)
PyObject* new_obj = PyLong_FromLong(100);
Py_XSETREF(obj, new_obj); // obj 引用计数 -1,然后 obj = new_obj
}2.3.2 New Reference vs Borrowed Reference
static PyObject* ref_demo(PyObject* self, PyObject* args) {
PyObject* list;
if (!PyArg_ParseTuple(args, "O!", &PyList_Type, &list))
return NULL;
// PyArg_ParseTuple 和大部分 Py*_New 函数返回 New Reference
PyObject* new_ref = PyList_GetItem(list, 0);
// 注意:PyList_GetItem 返回 Borrowed Reference!
// 不需要也不应该调用 Py_DECREF,除非显式增加了引用
Py_INCREF(new_ref); // 如果要将 Borrowed Reference 存储为长期引用,需要 INCREF
// ... 使用 new_ref ...
Py_DECREF(new_ref); // 使用完毕后释放
Py_RETURN_NONE;
}New Reference vs Borrowed Reference 判断规则:
| 函数 | 返回类型 |
|---|---|
PyLong_FromLong, PyUnicode_FromString 等 *_New 函数 | New Reference |
PyObject_GetItem, PyDict_GetItem | New Reference |
PyList_GetItem, PyTuple_GetItem | Borrowed Reference |
PyObject_GetAttr | New Reference |
PyImport_ImportModule | New Reference |
PyTuple_GetItem | Borrowed Reference |
2.3.3 循环引用处理
对于可能形成循环引用的容器对象(如自定义类型),需要实现 tp_traverse 和 tp_clear 方法:
typedef struct {
PyObject_HEAD
PyObject* other; // 可能与其他对象形成循环引用
} MyObject;
// 遍历被引用的对象
static int MyTraverse(MyObject* self, visitproc visit, void* arg) {
Py_VISIT(self->other); // 通知 GC 遍历此对象
return 0;
}
// 清除循环引用中的引用
static int MyClear(MyObject* self) {
Py_CLEAR(self->other); // Py_DECREF + 置 NULL
return 0;
}
// 在类型对象中注册
static PyTypeObject MyType = {
PyVarObject_HEAD_INIT(NULL, 0)
.tp_name = "my_module.MyObject",
.tp_basicsize = sizeof(MyObject),
.tp_flags = Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_GC,
.tp_traverse = (traverseproc)MyTraverse,
.tp_clear = (inquiry)MyClear,
// ...
};2.3.4 引用计数示例:完整函数
static PyObject* sum_list(PyObject* self, PyObject* args) {
PyObject* list;
Py_ssize_t n;
long total = 0;
if (!PyArg_ParseTuple(args, "O!", &PyList_Type, &list))
return NULL;
n = PyList_Size(list);
for (Py_ssize_t i = 0; i < n; i++) {
PyObject* item = PyList_GetItem(list, i); // Borrowed Reference
if (PyLong_Check(item)) {
long val = PyLong_AsLong(item);
if (val == -1 && PyErr_Occurred()) {
// PyLong_AsLong 出错时返回 -1 并设置异常
// 不需要 DECREF list,因为它是 Borrowed Reference(从 args 传入)
return NULL;
}
total += val;
}
}
// 返回新的 PyLong 对象(New Reference)
return PyLong_FromLong(total);
}2.4 C 扩展实例:高性能计算函数
2.4.1 数值数组求和计算
// array_sum.c
#include <Python.h>
#include <stddef.h>
static PyObject* array_sum_double(PyObject* self, PyObject* args) {
PyObject* input;
Py_buffer view;
// 使用缓冲区协议接收任何支持缓冲区的对象(如 bytes、bytearray、array.array、numpy array)
if (!PyArg_ParseTuple(args, "y*", &view)) {
return NULL;
}
if (view.ndim != 1) {
PyErr_SetString(PyExc_ValueError, "Only 1D arrays are supported");
PyBuffer_Release(&view);
return NULL;
}
double sum = 0.0;
Py_ssize_t n = view.len / view.itemsize;
// 根据格式化字符类型选择计算方式
if (view.format[0] == 'd' && view.itemsize == 8) {
// double 数组
double* data = (double*)view.buf;
for (Py_ssize_t i = 0; i < n; i++) {
sum += data[i];
}
} else if (view.format[0] == 'i' && view.itemsize == 4) {
// int 数组
int* data = (int*)view.buf;
for (Py_ssize_t i = 0; i < n; i++) {
sum += (double)data[i];
}
} else {
PyErr_SetString(PyExc_TypeError, "Unsupported array format");
PyBuffer_Release(&view);
return NULL;
}
PyBuffer_Release(&view);
return PyFloat_FromDouble(sum);
}
static PyMethodDef Methods[] = {
{"array_sum_double", array_sum_double, METH_VARARGS, "Sum array elements efficiently."},
{NULL, NULL, 0, NULL}
};
static struct PyModuleDef arraysum_module = {
PyModuleDef_HEAD_INIT,
"array_sum",
NULL,
-1,
Methods
};
PyMODINIT_FUNC PyInit_array_sum(void) {
return PyModule_Create(&arraysum_module);
}# setup.py
from setuptools import setup, Extension
ext = Extension(
"array_sum",
sources=["array_sum.c"],
)
setup(
name="array-sum",
version="1.0.0",
ext_modules=[ext],
)2.4.2 性能对比 Python vs C 扩展
# benchmark.py
import time
import array
# 准备数据
data = array.array('d', [float(i) for i in range(10_000_000)])
def sum_python(arr):
"""纯 Python 求和"""
total = 0.0
for x in arr:
total += x
return total
def sum_builtin(arr):
"""Python 内置 sum"""
return sum(arr)
# C 扩展求和
import array_sum
# 基准测试
for name, func in [("Python loop", sum_python), ("builtin sum", sum_builtin),
("C extension", array_sum.array_sum_double)]:
start = time.perf_counter()
result = func(data)
elapsed = time.perf_counter() - start
print(f"{name:20s}: {elapsed:.4f}s, result={result:.0f}")预期性能对比(10,000,000 个元素):
| 实现方式 | 耗时(秒) | 加速比 |
|---|---|---|
| 纯 Python for 循环 | ~2.500 | 1x |
| Python builtin sum | ~0.350 | ~7x |
| C 扩展 | ~0.025 | ~100x |
3. Cython
Cython 是一种带 C 类型声明的 Python 超集语言,可以编译为 C 扩展模块。
3.1 Cython 基础
3.1.1 .pyx 文件结构
# sum_cython.pyx
# cdef 声明 C 类型函数(只能在 Cython 内部调用)
cdef double _c_sum(double* arr, Py_ssize_t n) nogil:
cdef:
Py_ssize_t i
double total = 0.0
for i in range(n):
total += arr[i]
return total
# cpdef 声明 Python 和 C 双接口函数
cpdef double c_sum(list arr):
"""从 Python 列表中求和"""
cdef:
Py_ssize_t i, n = len(arr)
double total = 0.0
for i in range(n):
total += <double>arr[i]
return total
# def 声明纯 Python 接口函数
def c_sum_array(arr):
"""通过缓冲区协议接收任意数组类型"""
cdef:
double* data
Py_ssize_t n
double result
# 使用内存视图
cdef double[:] view = arr
n = view.shape[0]
data = &view[0]
with nogil: # 释放 GIL
result = _c_sum(data, n)
return result3.1.2 类型声明
# 基本类型声明
cdef int a = 42
cdef double b = 3.14
cdef char* s = "hello"
cdef Py_ssize_t n = 1000
# 指针声明
cdef int* ptr = &a
cdef double* dptr
# 数组声明(栈上分配)
cdef double arr[1000]
cdef int matrix[10][10]
# 结构体声明
cdef struct Point:
double x
double y
cdef Point p
p.x = 1.0
p.y = 2.0
# 联合体声明
cdef union Value:
int as_int
double as_double
char* as_string
# 函数指针类型
ctypedef double (*MathFunc)(double)
cdef double apply(double x, MathFunc f):
return f(x)3.1.3 cdef / cpdef / def 对比
| 关键字 | 调用方 | 可见性 | 性能 |
|---|---|---|---|
def | Python 代码 | 导出到 Python 命名空间 | 最慢(Python 函数调用) |
cdef | Cython 内部 | 仅在 Cython 内可见 | 最快(C 级别函数调用) |
cpdef | 两者 | 导出到 Python,内部使用 C | 接近 cdef 性能 |
3.2 与 Python 交互
3.2.1 Python 对象操作
# python_interop.pyx
from cpython cimport list, tuple, dict
def process_python_objects(py_list, py_dict):
cdef:
Py_ssize_t i, n
PyObject* key
PyObject* value
int result = 0
# 操作 Python 列表
if list.PyList_Check(py_list):
n = len(py_list)
for i in range(n):
item = py_list[i]
result += 1
# 操作 Python 字典
if dict.PyDict_Check(py_dict):
key = dict.PyDict_Next(py_dict, ...)
# ...
return result3.2.2 Python 调用 Cython
# main.py
import pyximport
pyximport.install() # 启用即时编译
# 直接导入 .pyx 文件
import sum_cython
result = sum_cython.c_sum([1.0, 2.0, 3.0])
print(result)3.2.3 pyximport 使用
# pyximport 适合开发和测试环境
import pyximport
pyximport.install(
language_level=3, # 使用 Python 3 语义
setup_args={
"include_dirs": [], # 额外的包含路径
"libraries": [], # 链接的库
},
reload_support=True, # 支持重新加载
)
import my_cython_module3.2.4 编译流程
方法 1:使用 setuptools
# setup.py
from setuptools import setup
from Cython.Build import cythonize
extensions = cythonize(
"sum_cython.pyx",
language_level=3, # Python 3 语义
annotate=True, # 生成 HTML 注释文件(显示类型优化情况)
compiler_directives={
"boundscheck": False, # 关闭边界检查
"wraparound": False, # 关闭负数索引
"nonecheck": False, # 关闭 None 检查
"cdivision": True, # 使用 C 除法
}
)
setup(
name="sum-cython",
ext_modules=extensions,
)# 构建
python setup.py build_ext --inplace方法 2:使用 pyproject.toml
[build-system]
requires = ["setuptools>=68.0", "wheel>=0.42", "Cython>=3.0"]
build-backend = "setuptools.build_meta"# setup.py
from setuptools import setup
from Cython.Build import cythonize
setup(
ext_modules=cythonize("sum_cython.pyx", language_level=3),
)3.3 性能优化
3.3.1 类型批注
# type_annotations.pyx
def compute(double[:] arr, int iterations):
"""显式类型批注可以极大提升性能"""
cdef:
Py_ssize_t i, j
Py_ssize_t n = arr.shape[0]
double total = 0.0
double temp
for j in range(iterations):
total = 0.0
for i in range(n):
total += arr[i]
return total3.3.2 释放 GIL(with nogil)
# nogil_demo.pyx
cdef double _pure_c_sum(double* arr, Py_ssize_t n) nogil:
"""nogil 函数:不能调用 Python API"""
cdef Py_ssize_t i
cdef double total = 0.0
for i in range(n):
total += arr[i]
return total
def parallel_sum(double[:] arr):
cdef double result
with nogil: # 在一个代码块中释放 GIL
result = _pure_c_sum(&arr[0], arr.shape[0])
return resultnogil 限制:
- 不能调用任何 Python/Cython
def或cpdef函数 - 不能操作 Python 对象(list、dict、str 等)
- 不能分配 Python 对象
- 不能触发 Python 异常
- 可以调用其他
cdef nogil函数
3.3.3 内存视图(memoryview)
# memoryview_demo.pyx
def process_array(double[:, :, :] arr3d):
"""三维数组处理"""
cdef:
Py_ssize_t i, j, k
Py_ssize_t d0 = arr3d.shape[0]
Py_ssize_t d1 = arr3d.shape[1]
Py_ssize_t d2 = arr3d.shape[2]
double total = 0.0
for i in range(d0):
for j in range(d1):
for k in range(d2):
total += arr3d[i, j, k]
return total
# 内存视图切片
def slice_sum(double[:] arr):
cdef double[:] sub = arr[100:200] # 不复制数据
cdef Py_ssize_t i
cdef double total = 0.0
for i in range(sub.shape[0]):
total += sub[i]
return total3.3.4 NumPy 集成
# numpy_integration.pyx
import numpy as np
cimport numpy as cnp
# 导入 NumPy 的 C API
cnp.import_array()
def sum_numpy_array(cnp.ndarray[double, ndim=1] arr):
"""直接操作 NumPy 数组内部数据"""
cdef:
Py_ssize_t i, n = arr.shape[0]
double* data = <double*>arr.data
double total = 0.0
for i in range(n):
total += data[i]
return total
def transform_numpy(cnp.ndarray[double, ndim=2] arr, double scale):
"""原地变换 NumPy 数组"""
cdef:
Py_ssize_t i, j
Py_ssize_t rows = arr.shape[0]
Py_ssize_t cols = arr.shape[1]
double* data = <double*>arr.data
for i in range(rows):
for j in range(cols):
data[i * cols + j] *= scale# 在编译配置中启用 NumPy
# setup.py
from setuptools import setup
from Cython.Build import cythonize
import numpy as np
setup(
ext_modules=cythonize(
"numpy_integration.pyx",
language_level=3,
),
include_dirs=[np.get_include()], # 包含 NumPy 头文件
)3.3.5 性能对比基准测试
# benchmark_cython.py
import time
import numpy as np
from sum_cython import c_sum, c_sum_array
from numpy_integration import sum_numpy_array
n = 10_000_000
data_list = [float(i) for i in range(n)]
data_np = np.arange(n, dtype=np.float64)
# 基准测试
results = []
# 1. 纯 Python sum
t0 = time.perf_counter()
r = sum(data_list)
t1 = time.perf_counter()
results.append(("Python builtin sum", t1 - t0))
# 2. Cython list sum
t0 = time.perf_counter()
r = c_sum(data_list)
t1 = time.perf_counter()
results.append(("Cython list sum", t1 - t0))
# 3. Cython memoryview sum
t0 = time.perf_counter()
r = c_sum_array(data_np)
t1 = time.perf_counter()
results.append(("Cython memoryview", t1 - t0))
# 4. Cython NumPy array sum
t0 = time.perf_counter()
r = sum_numpy_array(data_np)
t1 = time.perf_counter()
results.append(("Cython NumPy sum", t1 - t0))
# 5. NumPy builtin
t0 = time.perf_counter()
r = np.sum(data_np)
t1 = time.perf_counter()
results.append(("NumPy builtin sum", t1 - t0))
for name, elapsed in results:
print(f"{name:25s}: {elapsed:.4f}s")预期性能对比(10,000,000 个元素):
| 实现方式 | 相对性能 |
|---|---|
| Python builtin sum (list) | ~0.35s |
| Cython list sum (无类型) | ~0.30s |
| Cython list sum (有类型) | ~0.05s |
| Cython memoryview | ~0.04s |
| Cython NumPy sum | ~0.03s |
| NumPy builtin sum | ~0.02s |
3.4 编译指令
Cython 编译器指令优化:
| 指令 | 默认值 | 说明 |
|---|---|---|
boundscheck | True | 检查列表/数组越界访问 |
wraparound | True | 允许负数索引 |
nonecheck | True | 检查 None 值 |
cdivision | False | 使用 C 除法的截断行为 vs Python 的 floor 除法 |
cdivision_warnings | False | 对 C 除法发出警告 |
initializedcheck | True | 检查变量是否初始化 |
embedsignature | False | 在 docstring 中嵌入类型签名 |
# 在 .pyx 文件全局设置
# cython: boundscheck=False
# cython: wraparound=False
# cython: cdivision=True
# 或在使用 setuptools 时配置
# setup.py
from Cython.Build import cythonize
extensions = cythonize(
"module.pyx",
compiler_directives={
"boundscheck": False,
"wraparound": False,
}
)4. CFFI
CFFI(C Foreign Function Interface)是 Python 调用 C 库的另一种方式,提供比 ctypes 更优的性能和更完整的 C 类型支持。
4.1 CFFI 模式
CFFI 支持两种主要操作模式:
4.1.1 ABI 模式 vs API 模式
| 特性 | ABI 模式 | API 模式 |
|---|---|---|
| 依赖 | 运行时依赖共享库(.dll/.so) | 编译时需要头文件,运行时也可使用共享库 |
| 安全性 | 低,直接 ABI 调用 | 高,编译时验证类型兼容性 |
| 性能 | 略快(无编译开销) | 同 ABI 模式 |
| 使用便利性 | 简单 | 需要 C 编译器 |
| 推荐场景 | 快速原型、无头文件可用时 | 生产环境、正式项目 |
4.1.2 out-of-line vs in-line
| 特性 | in-line | out-of-line |
|---|---|---|
| 工作方式 | 运行时动态加载 C 代码 | 提前编译为 Python 扩展 |
| 加载时间 | 每次 import 时编译 | 只需一次的构建步骤 |
| 依赖 | FFI 构建时的编译器 | 用户安装时需要编译器 |
| 适合场景 | 开发、快速测试 | 发布、生产部署 |
4.2 调用 C 库完整示例
4.2.1 C 库头文件
// vec_ops.h
#ifndef VEC_OPS_H
#define VEC_OPS_H
typedef struct {
double x;
double y;
double z;
} Vector3D;
// 向量加法
Vector3D vec_add(const Vector3D* a, const Vector3D* b);
// 向量点积
double vec_dot(const Vector3D* a, const Vector3D* b);
// 向量标量乘法
Vector3D vec_scale(const Vector3D* v, double s);
// 向量长度
double vec_length(const Vector3D* v);
// 数值数组求和(错误码返回 -1)
int array_sum_double(const double* arr, int n, double* result);
#endif4.2.2 C 库实现
// vec_ops.c
#include "vec_ops.h"
#include <math.h>
Vector3D vec_add(const Vector3D* a, const Vector3D* b) {
Vector3D result;
result.x = a->x + b->x;
result.y = a->y + b->y;
result.z = a->z + b->z;
return result;
}
double vec_dot(const Vector3D* a, const Vector3D* b) {
return a->x * b->x + a->y * b->y + a->z * b->z;
}
Vector3D vec_scale(const Vector3D* v, double s) {
Vector3D result;
result.x = v->x * s;
result.y = v->y * s;
result.z = v->z * s;
return result;
}
double vec_length(const Vector3D* v) {
return sqrt(v->x * v->x + v->y * v->y + v->z * v->z);
}
int array_sum_double(const double* arr, int n, double* result) {
if (arr == NULL || result == NULL || n <= 0) {
return -1;
}
double sum = 0.0;
for (int i = 0; i < n; i++) {
sum += arr[i];
}
*result = sum;
return 0;
}4.2.3 CFFI API out-of-line 模式
# build_vec_ops.py
# 构建脚本:生成 C 扩展
from cffi import FFI
ffi = FFI()
# 声明 C 函数接口(从 .h 文件读取)
ffi.cdef("""
typedef struct {
double x;
double y;
double z;
} Vector3D;
Vector3D vec_add(const Vector3D* a, const Vector3D* b);
double vec_dot(const Vector3D* a, const Vector3D* b);
Vector3D vec_scale(const Vector3D* v, double s);
double vec_length(const Vector3D* v);
int array_sum_double(const double* arr, int n, double* result);
""")
# 设置源文件
ffi.set_source(
"_vec_ops",
'#include "vec_ops.h"', # 嵌入在生成的 C 代码中的头文件引用
sources=["vec_ops.c"], # 需要编译的 C 源文件
libraries=["m"], # 链接数学库
)
if __name__ == "__main__":
ffi.compile()# 使用 CFFI 扩展
from _vec_ops import ffi, lib
# 创建 Vector3D 结构体
v1 = ffi.new("Vector3D*", {"x": 1.0, "y": 2.0, "z": 3.0})
v2 = ffi.new("Vector3D*", {"x": 4.0, "y": 5.0, "z": 6.0})
# 调用 vec_add
result = lib.vec_add(v1, v2)
print(f"Add: ({result.x}, {result.y}, {result.z})")
# 调用 vec_dot
dot = lib.vec_dot(v1, v2)
print(f"Dot: {dot}")
# 调用 vec_length
length = lib.vec_length(v1)
print(f"Length: {length}")
# 调用 array_sum_double(数值数组操作)
import array
data = array.array('d', [1.0, 2.0, 3.0, 4.0, 5.0])
arr = ffi.cast("double*", ffi.from_buffer(data))
result = ffi.new("double*")
ret = lib.array_sum_double(arr, len(data), result)
if ret == 0:
print(f"Array sum: {result[0]}")
# 错误处理
ret = lib.array_sum_double(ffi.NULL, -1, result)
if ret == -1:
print("Error: invalid input")4.2.4 CFFI in-line 模式
# inline_cffi.py
from cffi import FFI
ffi = FFI()
# 声明 C 接口
ffi.cdef("""
int add(int a, int b);
double pi_value(void);
""")
# 内联 C 代码(运行时编译)
lib = ffi.verify("""
int add(int a, int b) {
return a + b;
}
double pi_value(void) {
return 3.14159265358979323846;
}
""")
print(lib.add(3, 4)) # 7
print(lib.pi_value()) # 3.1415926535897934.2.5 CFFI ABI 模式
# abi_cffi.py
from cffi import FFI
ffi = FFI()
# 声明 C 函数接口
ffi.cdef("""
typedef struct {
double x;
double y;
double z;
} Vector3D;
Vector3D vec_add(const Vector3D* a, const Vector3D* b);
double vec_dot(const Vector3D* a, const Vector3D* b);
""")
# ABI 模式直接加载共享库
lib = ffi.dlopen("vec_ops.dll") # Windows
# lib = ffi.dlopen("./libvec_ops.so") # Linux
v1 = ffi.new("Vector3D*", {"x": 1.0, "y": 2.0, "z": 3.0})
v2 = ffi.new("Vector3D*", {"x": 4.0, "y": 5.0, "z": 6.0})
result = lib.vec_add(v1, v2)
print(f"Result: ({result.x}, {result.y}, {result.z})")4.2.6 构建配置
# pyproject.toml
[build-system]
requires = ["setuptools>=68.0", "wheel>=0.42", "cffi>=1.16"]
build-backend = "setuptools.build_meta"# setup.py
from setuptools import setup
from cffi import FFI
ffi = FFI()
ffi.cdef("""
// ... 函数声明 ...
""")
ffi.set_source(
"_vec_ops",
'#include "vec_ops.h"',
sources=["vec_ops.c"],
libraries=["m"],
)
setup(
name="vec-ops",
version="1.0.0",
ext_modules=[ffi.distutils_extension()],
zip_safe=False,
)4.3 ctypes vs CFFI vs Cython vs C 扩展对比
| 方面 | ctypes | CFFI | Cython | C 扩展 (C API) |
|---|---|---|---|---|
| 性能 | 中(每次调用有装箱/拆箱开销) | 高(接近 C 扩展) | 高(编译为 C) | 最高(原生 C) |
| 需要 C 编译器 | 不需要 | API 模式需要/ABI 模式不需要 | 需要 | 需要 |
| 学习难度 | 低 | 中等 | 中等 | 高 |
| 开发效率 | 高 | 中等 | 中等 | 低 |
| 调试难度 | 低 | 中等 | 中等 | 高 |
| Python 类型支持 | 有限 | 好 | 完整 | 完整 |
| C 结构体布局 | 手动定义 | 自动推断 | 通过声明 | 手动操作 |
| 平台兼容性 | 跨平台 | 跨平台 | 需重新编译 | 需重新编译 |
| 文档生态 | 丰富 | 适中 | 丰富 | 官方文档完善 |
| 错误内存访问保护 | 无 | 部分(API 模式) | 部分 | 无 |
| 调用现有 C 库 | 简单 | 简单 | 需封装 | 需封装 |
| 编写新 C 代码 | 不适用 | 简单 | 自然 | 复杂 |
| NumPy 集成 | 通过缓冲区协议 | 通过缓冲区协议 | 原生支持 | 手动集成 |
选型建议:
- 快速调用现有 C 库:CFFI API out-of-line 模式(生产)/ ctypes(快速原型)
- 需要高性能 Python 代码:Cython
- 需要新写 C 代码且紧密集成:C 扩展(Python C API)
- 不想安装 C 编译器:ctypes 或 CFFI ABI 模式
5. PyPy 与 JIT
5.1 PyPy 介绍
PyPy 是 CPython 的 JIT 编译替代实现,专注于在保持兼容性的同时提供更优的性能。
5.1.1 JIT 编译
PyPy 的核心是即时编译(Just-In-Time Compilation)技术:
- 追踪式 JIT:PyPy 在运行时监控程序执行,识别热点代码路径(hot paths)
- 将这些热点路径编译为机器码,直接执行而非解释
- 自适应优化:根据运行时收集的类型信息生成特化代码
Python 源码
|
v
PyPy 解释器(RPython 实现)
|
v
JIT 编译器 <--- 运行时类型反馈
|
v
机器码(直接执行)5.1.2 Garbage Collection
PyPy 使用与 CPython 不同的垃圾回收策略:
| 特性 | CPython | PyPy |
|---|---|---|
| GC 策略 | 引用计数 + 循环检测 GC | 分代式 GC(generational GC) |
| 内存回收时机 | 对象引用计数归零时立即回收 | 在 GC 运行周期内批量回收 |
| 停止世界(STW) | 基本无(引用计数增量式) | 有短暂的 STW 停顿 |
| 内存碎片 | 较少 | 更多 |
| 大对象处理 | 直接 | 需要 GC 停顿 |
PyPy 的 GC 优势:
- 批量回收效率高,整体吞吐量好
- 无引用计数维护开销(减少缓存未命中)
- 对象移动(compaction)减少内存碎片
PyPy 的 GC 劣势:
- 有 GC 停顿(通常很短,但不适合硬实时场景)
- 内存占用通常高于 CPython
__del__方法的调用时机不可预测
5.1.3 CPython vs PyPy 兼容性
| 特性 | CPython | PyPy |
|---|---|---|
| Python 版本 | 3.12 / 3.13 | 3.10 兼容(PyPy 7.3.x) |
| C 扩展支持 | 原生 | 通过 cpyext(模拟 CPython C API) |
| C 扩展性能 | 原生 | cpyext 层有额外开销 |
| ctypes | 原生支持 | 原生支持(推荐替代 cpyext) |
| CFFI | 原生支持 | 原生支持(PyPy 推荐方式) |
| Cython | 原生支持 | 部分支持(某些特性受限) |
| asyncio | 完整支持 | 完整支持 |
| 第三方库 | 几乎所有 | 大多数纯 Python 库可用 |
5.2 PyPy 适用场景
推荐使用 PyPy 的场景:
- CPU 密集型纯 Python 代码(数值计算、字符串处理、模板渲染)
- 长时间运行的服务(JIT 预热后有性能优势)
- 大量小对象创建和销毁(分代 GC 更高效)
- 使用 CFFI / ctypes 调用 C 库的应用
不推荐使用 PyPy 的场景:
- 依赖大量 C 扩展且无 CFFI/ctypes 替代的代码
- 硬实时系统或对 GC 停顿敏感的应用
- 依赖
__del__方法确定性调用的代码 - 运行时间很短的任务(JIT 预热成本超过收益)
5.3 NumPy 兼容性(cpyext)
NumPy 在 PyPy 下的兼容情况:
# PyPy cpyext 对 NumPy 的支持
# 当前状态:PyPy 通过 cpyext 层支持 NumPy,但性能不如 CPython
import numpy as np
# 基本操作可用
arr = np.array([1.0, 2.0, 3.0])
print(arr.sum()) # 可用
print(arr.mean()) # 可用
print(arr.shape) # 可用
# 性能特征
# - 纯 Python 循环:PyPy JIT 比 CPython 快 2-10x
# - NumPy 操作:PyPy 比 CPython 慢(cpyext 层开销)
# - 混合场景:PyPy JIT 会跟踪 NumPy 操作,但整体不如 CPython
# 建议:如果代码以 NumPy 为主,使用 CPython
# 如果代码以 Python 逻辑为主(少量 NumPy),可以使用 PyPy不同场景下的性能对比(参考值):
| 场景 | CPython 3.12 | PyPy 7.3.x |
|---|---|---|
| 纯 Python 循环(CPU 密集型) | 1x | 2-10x 加速 |
| 字符串处理 | 1x | 2-5x 加速 |
| C 扩展(通过 cpyext) | 1x | 0.5-0.8x(性能损失) |
| NumPy 大数组操作 | 1x | 0.3-0.7x(性能损失) |
| 小型 Python 脚本(<1s) | 1x | 0.5-0.8x(JIT 预热开销) |
| 长时间运行的服务 | 1x | 1.5-3x 加速 |
使用 PyPy 时的最佳实践:
# 1. 优先使用 CFFI 而非 ctypes(PyPy 对 CFFI 有优化支持)
from cffi import FFI
# 2. 如果使用纯 Python,尽量保持风格简单
# PyPy JIT 对简单的循环和函数内联效果最好
# 3. 避免过多的 C 扩展调用
# 每次 C 扩展调用都需要通过 cpyext 层
# 4. 使用 timeit 实际测试性能
import timeit
# PyPy 的 JIT 需要预热,测试时使用足够的运行次数
# 5. 检查 PyPy 的兼容性
import sys
print(sys.version) # 确认运行在 PyPy 上5.4 PyPy 开发与部署
# 安装 PyPy
# Windows: 从 https://www.pypy.org/download.html 下载
# Linux (Ubuntu/Debian):
sudo apt install pypy3
# macOS:
brew install pypy3
# 创建虚拟环境
pypy3 -m venv pypy-venv
pypy-venv\Scripts\activate # Windows
source pypy-venv/bin/activate # Linux/macOS
# 安装纯 Python 包(无 C 扩展)
pip install requests flask jinja2
# 安装 CFFI 扩展
pip install cffi
# 条件判断代码运行环境
import sys
def is_pypy():
return hasattr(sys, "pypy_version_info")
if is_pypy():
print("Running on PyPy - JIT enabled")
else:
print("Running on CPython")PyPy 与 CPython 的选择决策流程:
是否需要调用大量 C 扩展?
├── 是 ──→ 依赖的 C 扩展是否有 CFFI 封装?
│ ├── 是(全部有) ──→ 可以尝试 PyPy
│ ├── 否(部分没有) ──→ 推荐 CPython
│ └── 大部分是 NumPy ──→ 推荐 CPython
├── 否 ──→ 代码类型?
├── CPU 密集型纯 Python ──→ PyPy(显著加速)
├── I/O 密集型 ──→ 两者均可(PyPy 略有优势)
├── 短期运行脚本 ──→ CPython(避免 JIT 预热)
└── 长时间运行服务 ──→ PyPy(预热后性能更好)