【Python 高性能编程·第八篇】C 扩展:用 ctypes 和 cffi 调用 C 库,零成本使用 BLAS/LAPACK
·
【Python 高性能编程·第八篇】C 扩展:用 ctypes 和 cffi 调用 C 库,零成本使用 BLAS/LAPACK
作者:技术博主 | 更新时间:2026-05-22 | 阅读时长:约 23 分钟
系列:Python 高性能编程(共 10 篇)
环境:Python 3.12,ctypes(内置),cffi,NumPy
标签:ctypescffiC扩展BLAS动态库FFI性能优化Python调用C

🔥 本篇目标:世界上最快的数值计算库是用 C/Fortran 写的——BLAS(矩阵运算)、LAPACK(线性代数)、FFTW(快速傅里叶变换)。这些库已经被几十年的工程优化打磨到接近硬件极限。Python 程序员不需要重新实现,只需要调用它们。
ctypes是 Python 内置的 C FFI(外部函数接口),cffi是更现代的替代方案。本篇从调用系统 C 库出发,到封装自己的 C 函数,再到直接调用 OpenBLAS 做矩阵乘法,全程不用写 Cython,不用配置编译环境。
系列进度
| 篇次 | 主题 | 状态 |
|---|---|---|
| 第一篇~第七篇 | 分析 / 向量化 / 内存 / 并发 / 多进程 / Cython / Numba | ✅ |
| 第八篇(本篇) | C 扩展:ctypes/cffi | — |
| 第九篇 | 数据结构与算法优化 | 即将发布 |
| 第十篇 | 综合实战:流水线提速 100× | 即将发布 |
目录
- 一、为什么直接调用 C 库
- 二、ctypes 基础:加载与调用
- 三、ctypes 数据类型与结构体
- 四、ctypes 与 NumPy 数组
- 五、cffi:更现代的 C 接口
- 六、调用 BLAS:直接用最快的矩阵运算
- 七、自己写 C 函数并调用
- 八、选择指南与实战总结
一、为什么直接调用 C 库
import ctypes
import numpy as np
import timeit
import platform
print("为什么要直接调用 C 库?")
print()
print(" 世界上最重要的数值计算库都是 C/Fortran 实现:")
print()
important_c_libs = [
("BLAS", "Basic Linear Algebra Subroutines", "矩阵乘法、向量运算的工业标准"),
("LAPACK", "Linear Algebra PACKage", "特征值、奇异值分解、线性方程组"),
("FFTW", "Fastest Fourier Transform in West", "最快的 FFT 实现"),
("OpenCV", "Open Computer Vision", "图像处理的标准库"),
("zlib", "压缩库", "Python 的 gzip 底层就用它"),
("libssl", "OpenSSL", "Python 的 ssl 模块底层"),
("sqlite3", "SQLite", "Python 的 sqlite3 模块底层"),
("libc", "C 标准库", "memcpy/qsort/sin/cos 等"),
]
print(f" {'库名':^10} {'全称':^36} {'用途':^28}")
print(" " + "─" * 78)
for lib, full, use in important_c_libs:
print(f" {lib:^10} {full:^36} {use:^28}")
print()
print(" Python 调用 C 库的方式:")
ways = [
("ctypes", "内置,无需安装", "简单,适合大多数场景"),
("cffi", "pip install cffi", "更现代,接近 C 语法"),
("Cython", "需要编译", "可以混写 Python/C(第六篇)"),
("SWIG", "需要编译", "自动生成绑定,适合大型 C++ 库"),
("pybind11", "需要编译", "C++ 库绑定的首选"),
("ctypesgen", "自动生成 ctypes", "从头文件自动生成绑定"),
]
print(f" {'方式':^12} {'安装':^18} {'特点':^28}")
print(" " + "─" * 62)
for way, install, feature in ways:
print(f" {way:^12} {install:^18} {feature:^28}")
print()
# 确认系统上可用的 C 库
system = platform.system()
print(f" 当前系统:{system}")
if system == "Linux":
print(" 标准 C 库路径:/lib/x86_64-linux-gnu/libc.so.6")
elif system == "Darwin":
print(" 标准 C 库路径:/usr/lib/libSystem.B.dylib")
elif system == "Windows":
print(" 标准 C 库路径:msvcrt.dll 或 ucrtbase.dll")
二、ctypes 基础:加载与调用
import ctypes
import ctypes.util
import platform
import time
print("\nctypes 基础:加载动态库并调用函数")
print()
# ── 加载标准 C 库 ─────────────────────────────────────────────
system = platform.system()
if system == "Linux":
libc = ctypes.CDLL("libc.so.6")
elif system == "Darwin":
libc = ctypes.CDLL("libSystem.B.dylib")
elif system == "Windows":
libc = ctypes.CDLL("msvcrt.dll")
print(f" 加载标准 C 库:{libc}")
print()
# ── 调用 C 的 printf ──────────────────────────────────────────
print(" 调用 C 的基础函数:")
print()
# strlen:字符串长度
strlen = libc.strlen
strlen.argtypes = [ctypes.c_char_p] # 参数类型:char*
strlen.restype = ctypes.c_size_t # 返回类型:size_t
result = strlen(b"Hello, World!")
print(f" strlen('Hello, World!') = {result}")
# abs:整数绝对值
abs_c = libc.abs
abs_c.argtypes = [ctypes.c_int]
abs_c.restype = ctypes.c_int
print(f" abs(-42) = {abs_c(-42)}")
# ── 数学函数:直接调用 C 的 sin/cos ──────────────────────────
import math
if system == "Linux":
libm = ctypes.CDLL("libm.so.6")
elif system == "Darwin":
libm = libc # macOS 的数学函数在 libSystem 里
elif system == "Windows":
libm = ctypes.CDLL("ucrtbase.dll")
try:
c_sin = libm.sin
c_sin.argtypes = [ctypes.c_double]
c_sin.restype = ctypes.c_double
c_sqrt = libm.sqrt
c_sqrt.argtypes = [ctypes.c_double]
c_sqrt.restype = ctypes.c_double
import timeit
# 性能对比:Python math vs ctypes C
x = 1.23456
t_math = timeit.timeit(lambda: math.sin(x), number=1_000_000)
t_ctypes = timeit.timeit(lambda: c_sin(x), number=1_000_000)
t_numpy = timeit.timeit(lambda: np.sin(x), number=1_000_000)
print()
print(f" sin(x) 性能对比(100万次):")
print(f" Python math.sin:{t_math*1000:.1f}ms")
print(f" ctypes C sin: {t_ctypes*1000:.1f}ms")
print(f" numpy np.sin: {t_numpy*1000:.1f}ms")
print()
print(" 注意:单次调用 ctypes 有 Python→C 的调用开销")
print(" ctypes 适合:批量大数据,而非逐个元素调用")
except Exception as e:
print(f" 数学库调用:{e}")
print()
# ── ctypes 的类型系统 ─────────────────────────────────────────
print(" ctypes 基本类型:")
ctypes_types = [
("ctypes.c_bool", "C bool", "Python bool"),
("ctypes.c_char", "C char", "Python bytes(1字节)"),
("ctypes.c_int", "C int", "Python int(32位)"),
("ctypes.c_long", "C long", "Python int(平台相关)"),
("ctypes.c_longlong", "C long long","Python int(64位)"),
("ctypes.c_uint", "C uint", "Python int(无符号32位)"),
("ctypes.c_float", "C float", "Python float(32位)"),
("ctypes.c_double", "C double", "Python float(64位)"),
("ctypes.c_char_p", "C char*", "Python bytes or None"),
("ctypes.c_void_p", "C void*", "Python int or None"),
]
print(f" {'ctypes 类型':^24} {'C 类型':^14} {'Python 类型':^20}")
print(" " + "─" * 62)
for ct, c, py in ctypes_types:
print(f" {ct:^24} {c:^14} {py:^20}")
三、ctypes 数据类型与结构体
import ctypes
print("\nctypes 数据类型与结构体:")
print()
# ── 基本类型的使用 ────────────────────────────────────────────
print(" 基本类型操作:")
print()
# 创建 ctypes 变量
c_int_val = ctypes.c_int(42)
c_float_val = ctypes.c_double(3.14)
c_str_val = ctypes.c_char_p(b"Hello")
print(f" c_int(42).value = {c_int_val.value}")
print(f" c_double(3.14).value = {c_float_val.value}")
print(f" c_char_p.value = {c_str_val.value}")
print()
# 指针
print(" 指针操作:")
x = ctypes.c_double(42.0)
ptr_x = ctypes.pointer(x) # 创建指向 x 的指针
print(f" x = {x.value}")
print(f" ptr_x.contents.value = {ptr_x.contents.value}")
ptr_x.contents.value = 99.0 # 通过指针修改
print(f" 修改后 x.value = {x.value}(通过指针写入)✓")
print()
# ── 数组 ─────────────────────────────────────────────────────
print(" C 数组:")
# 定义 C double 数组
DoubleArray5 = ctypes.c_double * 5
arr = DoubleArray5(1.1, 2.2, 3.3, 4.4, 5.5)
print(f" c_double * 5 数组:{list(arr)}")
print(f" arr[2] = {arr[2]}")
arr[0] = 99.9
print(f" arr[0] = 99.9 后:{list(arr)}")
print()
# ── 结构体 ────────────────────────────────────────────────────
print(" C 结构体(与 C 库交互必备):")
class Point2D(ctypes.Structure):
"""
等价的 C 结构体:
typedef struct {
double x;
double y;
} Point2D;
"""
_fields_ = [
("x", ctypes.c_double),
("y", ctypes.c_double),
]
class ComplexNumber(ctypes.Structure):
_fields_ = [
("real", ctypes.c_double),
("imag", ctypes.c_double),
]
p = Point2D(3.0, 4.0)
print(f" Point2D(3.0, 4.0): x={p.x}, y={p.y}")
print(f" size of Point2D: {ctypes.sizeof(Point2D)} bytes")
print(f" 欧氏距离: {(p.x**2 + p.y**2)**0.5}")
# 结构体数组
PointArray = Point2D * 3
points = PointArray(
Point2D(0.0, 0.0),
Point2D(1.0, 1.0),
Point2D(2.0, 2.0),
)
print(f" Point 数组:{[(p.x, p.y) for p in points]}")
print()
# ── 联合体(Union)────────────────────────────────────────────
print(" C 联合体(Union):")
class FloatBits(ctypes.Union):
"""查看 float 的二进制表示"""
_fields_ = [
("value", ctypes.c_float),
("bits", ctypes.c_uint32),
]
fb = FloatBits()
fb.value = 1.0
print(f" float 1.0 的二进制表示:0x{fb.bits:08X}")
fb.value = -0.0
print(f" float -0.0 的二进制: 0x{fb.bits:08X}")
fb.value = float('inf')
print(f" float inf 的二进制: 0x{fb.bits:08X}")
四、ctypes 与 NumPy 数组
import ctypes
import numpy as np
import timeit
print("\nctypes 与 NumPy 数组:高效传递大数组")
print()
print(" 关键:NumPy 数组的 ctypes 接口")
print()
# ── NumPy 数组转 ctypes 指针 ──────────────────────────────────
arr = np.array([1.0, 2.0, 3.0, 4.0, 5.0], dtype=np.float64)
# 方式1:arr.ctypes.data_as(最常用)
ptr_type = ctypes.POINTER(ctypes.c_double)
c_ptr = arr.ctypes.data_as(ptr_type)
print(f" arr = {arr}")
print(f" c_ptr 类型:{type(c_ptr)}")
print(f" c_ptr[0] = {c_ptr[0]},c_ptr[2] = {c_ptr[2]}")
print()
# 方式2:arr.ctypes.data(原始内存地址)
raw_addr = arr.ctypes.data
print(f" arr.ctypes.data(内存地址):0x{raw_addr:016X}")
print(f" ctypes.cast(addr, ptr) 恢复:{ctypes.cast(raw_addr, ptr_type)[0]}")
print()
# 方式3:np.ctypeslib(更便捷)
from numpy.ctypeslib import ndpointer
double_1d_array = ndpointer(dtype=np.float64, ndim=1, flags='C_CONTIGUOUS')
# 这个类型可以直接用在 argtypes 里!
# func.argtypes = [double_1d_array, ctypes.c_int]
# func(arr, len(arr)) # arr 自动转换为 C 指针
print(" np.ctypeslib.ndpointer:声明 NumPy 数组类型")
print(f" double_1d_array = {double_1d_array}")
print()
# ── 实战:把 NumPy 数组传给 C 函数 ───────────────────────────
print(" 实战:调用 C 的 qsort 对 NumPy 数组排序")
print()
# C 的 qsort 函数
if platform.system() == "Linux":
libc = ctypes.CDLL("libc.so.6")
elif platform.system() == "Darwin":
libc = ctypes.CDLL("libSystem.B.dylib")
else:
libc = ctypes.CDLL("msvcrt.dll")
# 定义比较函数(回调函数)
COMPARE_FUNC = ctypes.CFUNCTYPE(
ctypes.c_int, # 返回值:int
ctypes.c_void_p, # 参数1:const void*
ctypes.c_void_p, # 参数2:const void*
)
def double_compare(a_ptr, b_ptr):
"""比较两个 double 的大小(C 回调函数)"""
a = ctypes.cast(a_ptr, ctypes.POINTER(ctypes.c_double)).contents.value
b = ctypes.cast(b_ptr, ctypes.POINTER(ctypes.c_double)).contents.value
if a < b: return -1
if a > b: return 1
return 0
compare_func = COMPARE_FUNC(double_compare)
# 对 NumPy 数组排序
import numpy as np
data = np.array([5.0, 2.0, 8.0, 1.0, 9.0, 3.0], dtype=np.float64)
print(f" 排序前:{data}")
libc.qsort(
data.ctypes.data_as(ctypes.c_void_p), # 数组指针
ctypes.c_size_t(len(data)), # 元素数量
ctypes.c_size_t(data.itemsize), # 元素大小(8 bytes)
compare_func, # 比较函数
)
print(f" C qsort 后:{data} ✓")
print()
# ── 零拷贝:from_buffer 共享内存 ──────────────────────────────
print(" 零拷贝:ctypes 与 NumPy 共享内存")
print()
arr = np.arange(10, dtype=np.int32)
# 创建 ctypes 数组,与 NumPy 共享同一块内存
c_arr_type = ctypes.c_int32 * len(arr)
c_arr = c_arr_type.from_buffer(arr) # 零拷贝!
print(f" arr = {arr}")
print(f" c_arr = {list(c_arr)}")
c_arr[0] = 999 # 修改 ctypes 视图
print(f" 修改 c_arr[0]=999 后,arr = {arr} ← 共享内存!")
五、cffi:更现代的 C 接口
print("\ncffi:更现代的 C FFI")
print()
print(" cffi 的优势:")
print(" ① 直接写 C 头文件声明(比 ctypes 更接近 C 语法)")
print(" ② ABI 模式(不需要编译)和 API 模式(编译时检查)")
print(" ③ 更好的错误信息和类型检查")
print(" ④ 可以内联编写 C 代码(cdef + source)")
print()
try:
from cffi import FFI
ffi = FFI()
# ── ABI 模式:不需要编译,直接调用已有的动态库 ──────────
print(" cffi ABI 模式(不需要编译):")
print()
# 声明函数接口(直接写 C 声明!)
ffi.cdef("""
double sin(double x);
double cos(double x);
double sqrt(double x);
int abs(int x);
size_t strlen(const char *s);
void *memcpy(void *dest, const void *src, size_t n);
""")
# 加载动态库
import platform
if platform.system() == "Linux":
libm = ffi.dlopen("libm.so.6")
libc = ffi.dlopen("libc.so.6")
elif platform.system() == "Darwin":
libm = ffi.dlopen(None) # macOS:None = 当前进程(包含所有系统库)
libc = libm
else:
libm = ffi.dlopen("ucrtbase.dll")
libc = libm
# 调用(语法更自然!)
print(f" sin(3.14159/2) = {libm.sin(3.14159/2):.6f}")
print(f" sqrt(2.0) = {libm.sqrt(2.0):.6f}")
print(f" abs(-42) = {libc.abs(-42)}")
print(f" strlen('hello') = {libc.strlen(b'hello')}")
print()
# ── cffi 的 C 数组和指针 ──────────────────────────────────
print(" cffi 处理数组和指针:")
print()
# 分配 C 数组
c_arr = ffi.new("double[5]", [1.1, 2.2, 3.3, 4.4, 5.5])
print(f" double[5] = {list(c_arr)}")
print(f" c_arr[2] = {c_arr[2]}")
# 修改
c_arr[0] = 99.9
print(f" 修改后 c_arr[0] = {c_arr[0]}")
print()
# ── cffi 内联 C 代码(API 模式)──────────────────────────
print(" cffi API 模式(可以写 C 代码):")
print()
print('''
# ABI 模式:不需要编译器,直接运行
ffi.cdef("double sin(double x);")
lib = ffi.dlopen("libm.so.6")
result = lib.sin(1.0)
# API 模式:需要编译,但可以内联 C 代码
ffi.cdef("double fast_exp_approx(double x);")
ffi.set_source("_my_module", """
#include <math.h>
double fast_exp_approx(double x) {
// 快速近似 exp(仅演示)
return 1.0 + x + x*x/2.0 + x*x*x/6.0;
}
""")
ffi.compile() # 生成 _my_module.so
# 使用
from _my_module import ffi, lib
lib.fast_exp_approx(0.5)
''')
# ── cffi 与 NumPy 的集成 ─────────────────────────────────
print(" cffi 与 NumPy 的集成:")
import numpy as np
arr = np.array([1.0, 2.0, 3.0], dtype=np.float64)
# 获取 NumPy 数组的 cffi 指针
ptr = ffi.cast("double *", arr.ctypes.data)
print(f" NumPy 数组通过 cffi 访问:")
print(f" ptr[0]={ptr[0]}, ptr[1]={ptr[1]}, ptr[2]={ptr[2]}")
# 修改(共享内存!)
ptr[0] = 99.9
print(f" ptr[0]=99.9 后,arr[0]={arr[0]} ← 共享内存 ✓")
except ImportError:
print(" cffi 未安装,请运行:pip install cffi")
print()
print(" cffi 基本模式展示:")
print('''
from cffi import FFI
ffi = FFI()
# 声明(直接写 C 语法!)
ffi.cdef("""
double sin(double x);
double sqrt(double x);
int printf(const char *format, ...);
""")
# 加载库
import ctypes.util
libm = ffi.dlopen(ctypes.util.find_library("m"))
# 调用(比 ctypes 更自然)
print(libm.sin(1.0)) # 0.8414709848
''')
六、调用 BLAS:直接用最快的矩阵运算
import numpy as np
import ctypes
import timeit
import platform
print("\n调用 BLAS:最快的矩阵运算库")
print()
print(" BLAS(Basic Linear Algebra Subroutines):")
print(" 由 Intel MKL、OpenBLAS、ATLAS 等实现,针对 CPU 极度优化")
print(" NumPy/SciPy 底层就在调用 BLAS!")
print()
# ── 通过 NumPy 找到 BLAS ──────────────────────────────────────
print(" NumPy 使用的 BLAS 信息:")
try:
np_config = np.__config__.blas_opt_info
print(f" {np_config}")
except Exception:
print(" np.show_config() 可以看到 BLAS 配置")
print()
# ── 直接调用 BLAS 的 dgemm(矩阵乘法)────────────────────────
print(" 直接调用 BLAS dgemm(矩阵乘法):")
print()
print(" BLAS 命名规则:")
print(" d = double precision(双精度)")
print(" ge = general(通用矩阵)")
print(" mm = matrix-matrix multiply(矩阵×矩阵)")
print()
# 通过 scipy 调用 BLAS(最简单的方式)
try:
from scipy.linalg import blas as scipy_blas
# 创建测试矩阵
np.random.seed(42)
M, K, N = 512, 512, 512
A = np.random.randn(M, K).astype(np.float64)
B = np.random.randn(K, N).astype(np.float64)
# scipy 的 BLAS 接口
t_np = timeit.timeit(lambda: A @ B, number=20) / 20
t_blas = timeit.timeit(lambda: scipy_blas.dgemm(1.0, A, B), number=20) / 20
print(f" 矩阵乘法({M}×{K} × {K}×{N}):")
print(f" numpy A@B: {t_np*1000:.2f}ms")
print(f" scipy BLAS dgemm: {t_blas*1000:.2f}ms")
print(f" (两者底层都是 BLAS,速度应相近)")
print()
# 常用 BLAS 函数
print(" 常用 BLAS 函数(scipy.linalg.blas):")
blas_funcs = [
("ddot", "向量点积 a·b", "scipy_blas.ddot(a, b)"),
("daxpy", "向量更新 y = alpha*x + y", "scipy_blas.daxpy(x, y, a=alpha)"),
("dgemv", "矩阵-向量乘 y = A*x", "scipy_blas.dgemv(1.0, A, x)"),
("dgemm", "矩阵-矩阵乘 C = A*B", "scipy_blas.dgemm(1.0, A, B)"),
("dtrsv", "三角方程组 A*x = b", "scipy_blas.dtrsv(A, b)"),
("dnrm2", "向量 L2 范数 ||x||", "scipy_blas.dnrm2(x)"),
]
print(f" {'函数':^8} {'功能':^24} {'调用方式':^34}")
print(" " + "─" * 70)
for fn, desc, call in blas_funcs:
print(f" {fn:^8} {desc:^24} {call:^34}")
print()
# 向量运算性能对比
n = 10_000_000
a = np.random.randn(n)
b = np.random.randn(n)
t_np_dot = timeit.timeit(lambda: np.dot(a, b), number=50) / 50
t_blas_dot = timeit.timeit(lambda: scipy_blas.ddot(a, b), number=50) / 50
print(f" 向量点积(n={n:,}):")
print(f" np.dot: {t_np_dot*1000:.3f}ms")
print(f" BLAS ddot: {t_blas_dot*1000:.3f}ms")
print(f" (np.dot 内部调用 BLAS,速度相同)")
except ImportError:
print(" scipy 未安装。NumPy 本身也可以调用 BLAS:")
print(" np.dot, np.matmul (@) 内部都在调用 BLAS")
print()
print()
# ── 直接用 ctypes 调用 OpenBLAS ──────────────────────────────
print(" 直接用 ctypes 调用 OpenBLAS(更底层的控制):")
print()
print(" # 找到 OpenBLAS 动态库")
print(" import ctypes")
print(" openblas = ctypes.CDLL('libopenblas.so')")
print()
print(" # 声明 dgemm 接口")
print(" openblas.dgemm_.argtypes = [")
print(" ctypes.c_char_p, # transa")
print(" ctypes.c_char_p, # transb")
print(" ctypes.POINTER(ctypes.c_int), # M")
print(" ctypes.POINTER(ctypes.c_int), # N")
print(" ctypes.POINTER(ctypes.c_int), # K")
print(" ctypes.POINTER(ctypes.c_double), # alpha")
print(" ctypes.c_void_p, # A")
print(" ctypes.POINTER(ctypes.c_int), # lda")
print(" ...")
print(" ]")
print()
print(" # 注意:BLAS Fortran 接口用列优先(Fortran order)")
print(" # NumPy 默认是行优先(C order),需要处理转置")
七、自己写 C 函数并调用
import ctypes
import numpy as np
import tempfile
import os
import subprocess
import timeit
print("\n自己写 C 函数并从 Python 调用:")
print()
# ── 流程:写 C → 编译 → ctypes 加载 ─────────────────────────
print(" 完整流程:")
print()
print(" 1. 写 C 源文件(.c)")
print(" 2. 编译为动态库(.so / .dll)")
print(" 3. ctypes.CDLL 加载")
print(" 4. 声明参数/返回类型")
print(" 5. 调用函数")
print()
# ── 示例 C 代码 ───────────────────────────────────────────────
c_source = '''
#include <stdio.h>
#include <math.h>
#include <stdlib.h>
// 1. 简单函数:两数相加
double add_doubles(double a, double b) {
return a + b;
}
// 2. 数组操作:就地平方(修改原数组)
void square_array_inplace(double *arr, int n) {
for (int i = 0; i < n; i++) {
arr[i] = arr[i] * arr[i];
}
}
// 3. 返回新数组:调用方负责 free()
double* sqrt_array(const double *arr, int n) {
double *result = (double*)malloc(n * sizeof(double));
if (result == NULL) return NULL;
for (int i = 0; i < n; i++) {
result[i] = sqrt(arr[i]);
}
return result;
}
// 4. 快速排序(演示 C 实现的优势)
void fast_sort(double *arr, int n) {
// 用标准库 qsort
int compare(const void *a, const void *b) {
double da = *(double*)a;
double db = *(double*)b;
return (da > db) - (da < db);
}
qsort(arr, n, sizeof(double), compare);
}
// 5. 结构体操作
typedef struct {
double x;
double y;
} Vec2D;
double vec2d_length(Vec2D v) {
return sqrt(v.x * v.x + v.y * v.y);
}
Vec2D vec2d_add(Vec2D a, Vec2D b) {
Vec2D result = {a.x + b.x, a.y + b.y};
return result;
}
'''
print(" 示例 C 代码(my_lib.c):")
print()
# 打印关键部分
for line in c_source.split('\n')[2:18]:
print(f" {line}")
print(" ...")
print()
# ── 编译(如果有 gcc)────────────────────────────────────────
print(" 编译命令:")
print()
system = platform.system()
if system == "Linux":
compile_cmd = "gcc -shared -fPIC -O2 -o my_lib.so my_lib.c -lm"
elif system == "Darwin":
compile_cmd = "gcc -shared -fPIC -O2 -o my_lib.dylib my_lib.c -lm"
elif system == "Windows":
compile_cmd = "cl /LD /O2 my_lib.c /link /OUT:my_lib.dll"
print(f" {compile_cmd}")
print()
print(" 参数说明:")
compile_params = [
("-shared", "生成共享库(.so)而非可执行文件"),
("-fPIC", "位置无关代码(共享库必须)"),
("-O2", "优化级别2(-O3 更激进)"),
("-o my_lib.so", "输出文件名"),
("-lm", "链接数学库(for sqrt 等)"),
]
for param, desc in compile_params:
print(f" {param:16s}: {desc}")
print()
# ── ctypes 加载和使用 ─────────────────────────────────────────
print(" ctypes 加载自定义库:")
print()
print('''
import ctypes
import numpy as np
# 加载
lib = ctypes.CDLL("./my_lib.so") # Linux/Mac
# lib = ctypes.WinDLL("my_lib.dll") # Windows
# ① 简单函数
lib.add_doubles.argtypes = [ctypes.c_double, ctypes.c_double]
lib.add_doubles.restype = ctypes.c_double
result = lib.add_doubles(3.14, 2.71) # 5.85
# ② 数组就地操作(直接传 NumPy 指针,零拷贝)
lib.square_array_inplace.argtypes = [
ctypes.POINTER(ctypes.c_double),
ctypes.c_int
]
arr = np.array([1.0, 2.0, 3.0, 4.0])
lib.square_array_inplace(
arr.ctypes.data_as(ctypes.POINTER(ctypes.c_double)),
len(arr)
)
# arr 现在是 [1.0, 4.0, 9.0, 16.0]
# ③ 返回指针(需要手动释放内存!)
lib.sqrt_array.argtypes = [ctypes.POINTER(ctypes.c_double), ctypes.c_int]
lib.sqrt_array.restype = ctypes.POINTER(ctypes.c_double)
lib.free.argtypes = [ctypes.c_void_p] # 用 C 的 free 释放
arr_in = np.array([4.0, 9.0, 16.0, 25.0])
ptr_out = lib.sqrt_array(
arr_in.ctypes.data_as(ctypes.POINTER(ctypes.c_double)),
len(arr_in)
)
result = np.ctypeslib.as_array(ptr_out, shape=(len(arr_in),)).copy()
lib.free(ptr_out) # 务必释放!否则内存泄漏
# ④ 结构体
class Vec2D(ctypes.Structure):
_fields_ = [("x", ctypes.c_double), ("y", ctypes.c_double)]
lib.vec2d_length.argtypes = [Vec2D]
lib.vec2d_length.restype = ctypes.c_double
v = Vec2D(3.0, 4.0)
length = lib.vec2d_length(v) # 5.0
''')
# ── 临时演示:用 numpy 模拟等效性能 ──────────────────────────
print(" 性能演示(等效 C 函数的性能):")
print()
N = 5_000_000
arr = np.random.rand(N)
def square_python(arr):
for i in range(len(arr)):
arr[i] = arr[i] * arr[i]
# 模拟 C 就地平方的速度(NumPy **= 2 等效)
t_py = timeit.timeit(lambda: [x*x for x in arr[:10000]], number=100) / 100
t_np = timeit.timeit(lambda: arr ** 2, number=100) / 100
print(f" 数组平方(n={N:,}):")
print(f" Python 循环(n=1万):{t_py*1000:.2f}ms(外推100万:{t_py*100:.0f}ms)")
print(f" NumPy/C 就地: {t_np*1000:.2f}ms")
print(f" 自定义 C 函数速度类似 NumPy(直接内存操作)")
八、选择指南与实战总结
import timeit
import numpy as np
print("\n选择指南:ctypes vs cffi vs Cython vs Numba")
print()
print(" ┌─────────────────────────────────────────────────────────┐")
print(" │ 需要调用 C/C++ 库? │")
print(" │ ├── 库已存在(.so/.dll) │")
print(" │ │ ├── 简单接口(<10个函数)→ ctypes(内置,无需安装) │")
print(" │ │ └── 复杂接口(大型库) → cffi 或 ctypesgen │")
print(" │ └── 需要写新的 C 代码 │")
print(" │ ├── 数值计算循环 → Numba @njit(最简单) │")
print(" │ ├── 复杂算法 → Cython(完整 C 控制) │")
print(" │ └── C++ 库 → pybind11(最适合 C++) │")
print(" └─────────────────────────────────────────────────────────┘")
print()
# 各方案对比
comparison = [
("ctypes", "内置", "中", "中", "✅ 已有的 C 库,简单接口"),
("cffi", "pip", "低", "高", "✅ 已有 C 库,更好的 C 语法"),
("Numba", "pip", "低", "高", "✅ 数值计算,不想改代码"),
("Cython", "编译", "高", "极高", "✅ 复杂算法,发布扩展包"),
("pybind11", "编译", "高", "极高", "✅ C++ 库绑定"),
("SWIG", "编译", "中", "高", "✅ 大型 C/C++ 库自动绑定"),
]
print(f" {'方案':^10} {'安装':^8} {'学习曲线':^10} {'性能':^8} {'最适合':^32}")
print(" " + "─" * 74)
for name, install, curve, perf, best in comparison:
print(f" {name:^10} {install:^8} {curve:^10} {perf:^8} {best:^32}")
print()
print("=" * 65)
print(" 本系列加速方案完整速查(单核 CPU 为主):")
print("=" * 65)
print()
speedup_table = [
("Python 列表循环", "1×", "基准"),
("NumPy 向量化(第二篇)", "10-1000×", "大多数数值计算首选"),
("Numba @njit(第七篇)", "50-500×", "无法向量化的循环"),
("Cython 内存视图(第六篇)","100-1000×","复杂算法,极限性能"),
("ctypes/cffi(本篇)", "100-1000×","调用已有 C 库"),
("直接 BLAS(本篇)", "接近硬件", "矩阵/向量运算极限"),
("多进程(第五篇)", "N×核数", "CPU 密集,真正并行"),
("asyncio(第四篇)", "IO并发", "网络/文件 IO"),
("Numba CUDA(第七篇)", "100-1000×","GPU,大规模数据并行"),
]
print(f" {'方案':^28} {'加速':^12} {'适用场景':^24}")
print(" " + "─" * 68)
for method, speedup, use_case in speedup_table:
print(f" {method:^28} {speedup:^12} {use_case:^24}")
print()
print(" 终极建议:")
final_tips = [
"先 profile(第一篇),找真正的瓶颈",
"能向量化就先向量化(第二篇 NumPy)",
"向量化不够用 → Numba @njit(最省力)",
"需要极限性能 → Cython(最快)",
"调用 C 库 → ctypes/cffi(最灵活)",
"矩阵运算 → 直接用 NumPy(底层 BLAS)",
"IO 密集 → asyncio(第四篇)",
"CPU 密集且多核 → multiprocessing(第五篇)",
]
for i, tip in enumerate(final_tips, 1):
print(f" {i}. {tip}")
总结
ctypes vs cffi 选择:
| 特性 | ctypes | cffi |
|---|---|---|
| 安装 | 内置,无需安装 | pip install cffi |
| 声明语法 | Python 风格 | C 语法(直接写 C 头文件) |
| 性能 | 相同 | 相同 |
| 错误信息 | 较差 | 较好 |
| 内联 C 代码 | ❌ | ✅(API 模式) |
| 推荐场景 | 简单调用,快速原型 | 复杂接口,生产代码 |
调用 C 库的三步法:
- 加载库:
ctypes.CDLL("libm.so.6")/ffi.dlopen("libm.so.6") - 声明类型:
func.argtypes = [...],func.restype = ... - 传递数组:
arr.ctypes.data_as(POINTER(c_double))零拷贝
下一篇预告:数据结构与算法优化——Python 标准库中 collections.deque、heapq、bisect、sortedcontainers 的正确打开方式,以及常见算法从 O ( n 2 ) O(n^2) O(n2) 优化到 O ( n log n ) O(n \log n) O(nlogn) 的实战案例。
💬 你有直接用 ctypes 调用 C 库解决过什么问题?有没有踩过内存泄漏的坑? 欢迎评论区分享!
🙏 如果这篇帮到你,点赞 + 收藏,系列持续更新!
本文为原创技术分享。代码在 Python 3.12 + ctypes(内置)+ cffi 下验证。最后更新:2026-05-22
更多推荐


所有评论(0)