使用 Cython 开发 Python 扩展:完整教程
Cython 是 Python 的一个超集,允许开发者编写 Python 代码并将其编译为 C 扩展。它是 Python 生态中最成熟的扩展开发方案之一,广泛应用于科学计算和性能关键型应用。
为什么选择 Cython?
- 渐进式优化:可以在不修改原有 Python 代码的情况下逐步添加类型声明
- 无缝集成:支持大多数 Python 模块和 C/C++ 库
- 易于学习:语法与 Python 非常相似,学习曲线平缓
- 广泛应用:NumPy、SciPy 等著名库都使用 Cython 开发
环境准备
安装 Cython:
pip install cython项目结构
myextension/
├── setup.py # 构建配置
├── mymodule.pyx # Cython 源码
└── test_module.py # 测试脚本基础实现
1. 简单函数
创建 mymodule.pyx 文件:
# mymodule.pyx
def fibonacci(int n):
"""计算斐波那契数列"""
if n == 0:
return 0
cdef int a = 0
cdef int b = 1
cdef int i
for i in range(1, n):
a, b = b, a + b
return b2. 类型声明
Cython 的核心优势是类型声明:
# 静态类型声明
cdef int a = 10
cdef double b = 3.14
cdef str name = "hello"
cdef list numbers = [1, 2, 3]
cdef dict data = {"key": "value"}
# C 类型
cdef float* pointer
cdef struct Point:
double x
double y
# Python 对象(使用 Python 语法)
cdef object result3. 函数参数类型
# 指定参数类型
def add(int a, int b):
return a + b
# 带默认值
def greet(str name="World"):
return f"Hello, {name}!"
# 可选参数
def process(data, int mode=-1):
pass
# *args 和 **kwargs
def flexible(*args, **kwargs):
print(args, kwargs)4. 类定义
cdef class Counter:
cdef int count
cdef str name
def __init__(self, str name="default", int initial_value=0):
self.name = name
self.count = initial_value
def increment(self, int amount=1):
self.count += amount
return self.count
# 属性访问
property count:
def __get__(self):
return self.count
def __set__(self, int value):
self.count = value
# 字符串表示
def __str__(self):
return f"Counter({self.name}, {self.count})"
# 运算符重载
def __add__(self, Counter other):
return Counter(self.name + "+" + other.name,
self.count + other.count)5. 扩展类型(cdef class)
cdef class Point:
cdef public double x
cdef public double y
def __init__(self, double x, double y):
self.x = x
self.y = y
cpdef double distance(self, Point other):
"""cpdef 允许从 Python 调用,同时支持 C 调用"""
cdef double dx = self.x - other.x
cdef double dy = self.y - other.y
return (dx * dx + dy * dy) ** 0.56. 迭代器
cdef class FibonacciIterator:
cdef int current
cdef int next
cdef int max_iterations
cdef int iteration
def __init__(self, int max_iterations):
self.current = 0
self.next = 1
self.max_iterations = max_iterations
self.iteration = 0
def __iter__(self):
return self
def __next__(self):
if self.iteration >= self.max_iterations:
raise StopIteration
result = self.current
self.current, self.next = self.next, self.current + self.next
self.iteration += 1
return result7. 与 C/C++ 集成
声明 C 函数
cdef extern from "math.h":
double sin(double x)
double cos(double x)
double sqrt(double x)
# 使用 C 函数
def calculate(double x):
return sin(x) * sqrt(x)使用 C 结构体
cdef extern from "stdlib.h":
struct Point:
double x
double y
Point* malloc(size_t size)
void free(Point* ptr)8. 内存视图(Memory Views)
高效处理 NumPy 数组:
import numpy as np
cimport numpy as np
cimport cython
# 声明内存视图类型
cdef double[:] view_1d
cdef double[:, :] view_2d
def sum_array(double[:] arr):
"""使用内存视图处理数组"""
cdef double total = 0.0
cdef int i
for i in range(arr.shape[0]):
total += arr[i]
return total
@cython.boundscheck(False)
@cython.wraparound(False)
def fast_sum(double[:] arr):
"""禁用边界检查以提高性能"""
cdef double total = 0.0
cdef Py_ssize_t i
cdef Py_ssize_t n = arr.shape[0]
for i in range(n):
total += arr[i]
return total9. GIL 管理
释放全局解释器锁以提高并发性能:
from cython cimport floating
cimport numpy as np
cimport cython
@cython.boundscheck(False)
@cython.wraparound(False)
def parallel_sum(double[:] arr):
cdef double result = 0.0
cdef Py_ssize_t i
# 释放 GIL
with nogil:
for i in range(arr.shape[0]):
with gil:
# 临时获取 GIL 进行 Python 操作
result += arr[i]
return result构建配置
setup.py
from setuptools import setup
from Cython.Build import cythonize
from setuptools.extension import Extension
extensions = [
Extension("mymodule",
sources=["mymodule.pyx"],
extra_compile_args=["-O3", "-ffast-math"],
language="c++")
]
setup(
name="mymodule",
ext_modules=cythonize(
extensions,
compiler_directives={
'language_level': 3,
'boundscheck': False,
'wraparound': False,
'initializedcheck': False,
}
)
)pyproject.toml
[build-system]
requires = ["setuptools", "cython"]
build-backend = "setuptools.build_meta"
[project]
name = "mymodule"
requires-python = ">=3.8"
[tool.cython]
language_level = 3
boundscheck = false
wraparound = false构建和安装
# 开发模式安装
pip install -e .
# 或使用 setup.py
python setup.py build_ext --inplace
# 完整构建
python setup.py sdist bdist_wheel性能优化技巧
1. 使用 cdef 代替 def
# 慢:使用 Python 调用约定
def slow_function(int x):
return x * 2
# 快:使用 C 调用约定
cdef fast_function(int x):
return x * 22. 使用 cpdef
# 可以在 Python 和 C 中调用
cpdef double calculate(double x, double y):
return x * y + x / y3. 禁用边界检查
@cython.boundscheck(False)
@cython.wraparound(False)
def optimized_loop(double[:] arr):
# 快速循环代码
pass4. 使用 Typed Memoryview
def process_arrays(double[:, :] arr1, double[:, :] arr2):
cdef int i, j
for i in range(arr1.shape[0]):
for j in range(arr1.shape[1]):
arr1[i, j] += arr2[i, j]实际应用示例
数值计算
import numpy as np
cimport numpy as np
cimport cython
@cython.boundscheck(False)
@cython.wraparound(False)
def matrix_multiply(np.ndarray[np.float64_t, ndim=2] A,
np.ndarray[np.float64_t, ndim=2] B):
cdef int n = A.shape[0]
cdef int m = A.shape[1]
cdef int p = B.shape[1]
cdef np.ndarray[np.float64_t, ndim2] C = np.zeros((n, p))
cdef int i, j, k
for i in range(n):
for j in range(p):
for k in range(m):
C[i, j] += A[i, k] * B[k, j]
return C字符串处理
def process_strings(list strings):
cdef list result = []
cdef str s
for s in strings:
result.append(s.upper())
return result与 PyO3 对比
| 特性 | Cython | PyO3 (Rust) |
|---|---|---|
| 学习曲线 | 低 | 中等 |
| 性能 | 高 | 极高 |
| 内存安全 | 手动管理 | 编译时保证 |
| 类型系统 | 渐进式 | 静态类型 |
| C/C++ 集成 | 优秀 | 良好 |
| 生态 | 成熟稳定 | 快速成长 |
最佳实践
- 渐进式优化:先用纯 Python 编写,然后逐步添加类型声明
- 性能测试:使用
timeit定期测试性能提升 - 保持兼容性:确保代码在无类型声明时也能正常工作
- 使用编译器指令:针对具体场景优化编译选项
总结
Cython 是一个成熟且强大的工具,特别适合:
- 优化现有 Python 代码
- 封装 C/C++ 库
- 需要与 NumPy 深度集成的项目
它的渐进式优化特性和低学习曲线使其成为 Python 扩展开发的热门选择。