ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

Python C 扩展与 ffi 实战2026版:从 ctypes 到 Cython 混合编程的完整指南

Python C 扩展与 ffi 实战2026版:从 ctypes 到 Cython 混合编程的完整指南 Python C 扩展与 ffi 实战2026版:从 ctypes 到 Cython 混合编程的完整指南本文是 Python 高级应用系列的第 5 篇(终篇)。当纯 Python 性能不够时,C 扩展是最后的杀手锏。本文系统讲解 ctypes、cffi、Cython、C API 四种方案的实战对比。 — ## 一、为什么需要 C 扩展? | 场景 | 纯 Python 的问题 | C 扩展的解决方案 || :— | :— | :— ||数值计算| GIL 限制 解释器开销 | 直接操作 C æ•°ç»„ï¼Œæ— GIL ||调用系统库| 需要封è£å±‚ | 直接调用 C å±äº«åº“ ||嵌å¥è§£é‡Šå™¨| æ— æ³•åµŒå¥ C 程序 | Python C API 双向交互 ||性能瓶颈| å¾ªçŽ¯æ¢ | ç¼–è¯‘ä¸ºæœºå™¨ç  ||å†å­˜æŽ§åˆ¶| GC 不可控 | 手动å†å­˜ç®¡ç† | ### 四种方案对比 | 方案 | 难度 | 性能 | 灵活性 | 适用场景 || :—: | :—: | :—: | :—: | :— ||ctypes| 低 | 中 | 高 | 调用现有 C 库 ||cffi| 中 | 中高 | 高 | 调用 C 库 å†è” C ||Cython| 中 | 极高 | 中 | 编写高性能扩展 ||C API| 高 | 极高 | 极高 | 底层开发、嵌å¥è§£é‡Šå™¨ | é€‰æ‹©å†³ç­–æ ‘ï¼šâ”œâ”€â”€ 只需调用现有 .so/.dll → ctypes(最快上手)├── 需要å†è” C ä»£ç  → cffi(API 模式)├── 需要重写 Python 算法 → Cython(最佳平衡)└── 需要精细控制 Python 解释器 → C API(终极方案)— ## 二、ctypes:最简单的 C 调用 ### 2.1 è°ƒç”¨æ ‡å‡† C 库 pythonimport ctypesimport ctypes.utilimport time 调用 C æ ‡å‡†åº“ # åŠ è½½ libclibc ctypes.CDLL(ctypes.util.find_library(‘c’))# 调用 printflibc.printf(bHello from C! %d\n, 42)调用 abslibc.abs.restype ctypes.c_intlibc.abs.argtypes [ctypes.c_int]print(fabs(-42) {libc.abs(-42)}“)# 调用 strlenlibc.strlen.restype ctypes.c_size_tlibc.strlen.argtypes [ctypes.c_char_p]print(fstrlen(‘hello’) {libc.strlen(b’hello’)}”)# 调用 qsort(排序)CompareFunc ctypes.CFUNCTYPE( ctypes.c_int, # 返回值 ctypes.POINTER(ctypes.c_int), # 参数1 ctypes.POINTER(ctypes.c_int), # 参数2)def py_compare(a, b): “”“Python 回调函数”“”a_val a[0] b_val b[0] return a_val - b_valcmp_func CompareFunc(py_compare)准备数据arr (ctypes.c_int * 10)(5, 3, 8, 1, 9, 2, 7, 4, 6, 0)print(f排序前: {list(arr)}“)libc.qsort(arr, len(arr), ctypes.sizeof(ctypes.c_int), cmp_func)print(f排序后: {list(arr)}”)# 输出:# Hello from C! 42# abs(-42) 42# strlen(‘hello’) 5# 排序前: [5, 3, 8, 1, 9, 2, 7, 4, 6, 0]# 排序后: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]### 2.2 定义 C 结构体 pythonimport ctypes C ç»“æž„ä½“æ˜ å°„ class Point(ctypes.Structure):fields [ (“x”, ctypes.c_double), (“y”, ctypes.c_double), ]class Rectangle(ctypes.Structure):fields [ (“top_left”, Point), (“bottom_right”, Point), ] def area(self): width abs(self.bottom_right.x - self.top_left.x) height abs(self.bottom_right.yself.top_left.y) return width * height defrepr(self): return fRectangle({self.top_left.x}, {self.top_left.y}, {self.bottom_right.x}, {self.bottom_right.y})“# 创建rect Rectangle( Point(0.0, 0.0), Point(10.0, 5.0))print(f矩形: {rect}”)print(f面积: {rect.area()})# 调用自定义 C 库 # 假设有 mylib.c:# c# #include stdio.h# # typedef struct {# double x, y;# } Point;# # double distance(Point a, Point b) {# double dx a.x - b.x;# double dy a.yb.y;# return sqrt(dx * dx dy * dy);# }# 编译: gcc -shared -o libmylib.so mylib.c# åŠ è½½# mylib ctypes.CDLL(‘./libmylib.so’)mylib.distance.restype ctypes.c_double# mylib.distance.argtypes [Point, Point]# # p1 Point(0.0, 0.0)p2 Point(3.0, 4.0)dist mylib.distance(p1, p2)print(f距离: {dist}) # 5.0nbsp;### 2.3 ctypes 实战:调用 OpenSSLnbsp;python import ctypes import os # åŠ è½½ OpenSSLssl_lib ctypes.CDLL(libssl.so)crypto_lib ctypes.CDLL(libcrypto.so) # SHA256 哈希crypto_lib.SHA256.argtypes [ ctypes.c_char_p, # 数据 ctypes.c_size_t, # 长度 ctypes.c_char_p, # 输出缓冲区]crypto_lib.SHA256.restype ctypes.c_char_pdef sha256_hex(data: str) - str: 使用 OpenSSL 计算 SHA256 data_bytes data.encode(utf-8) output ctypes.create_string_buffer(32) crypto_lib.SHA256(data_bytes, len(data_bytes), output) return output.raw.hex()print(fSHA256(hello) {sha256_hex(hello)})# 输出:SHA256(hello) 2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824# 对比 Python æ ‡å‡†åº“import hashlibpy_hash hashlib.sha256(bhello).hexdigest()print(fPython SHA256 {py_hash}) print(f结果一致: {sha256_hex(hello) py_hash})# 输出:结果一致: True nbsp;---nbsp;## 三、cffi:更现代的 C 接口nbsp;### 3.1 ABI 模式(类似 ctypes 但更好用)nbsp;python from cffi import FFIffi FFI() # 声明 C 类型和函数签名ffi.cdef( int abs(int); size_t strlen(const char *); void qsort(void *base, size_t nmemb, size_t size, int (*compar)(const void *, const void *)); double sqrt(double); )# åŠ è½½æ ‡å‡†åº“libc ffi.dlopen(None) # 调用print(fabs(-42) {libc.abs(-9)})print(fstrlen(hello) {libc.strlen(bhello)})print(fsqrt(2) {libc.sqrt(2.0)})# qsortarr ffi.new(int[], [5, 3, 8, 1, 9, 2, 7, 4, 6, 0])ffi.callback(int(const void *, const void *))def compare(a, b): return ffi.cast(int *, a)[0] - ffi.cast(int *, b)[0]libc.qsort(arr, 10, ffi.sizeof(int), compare)print(f排序后: {list(arr)})# 输出:# abs(-42) 9# strlen(hello) 5# sqrt(2) 1.4142135623730951# 排序后: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] nbsp;### 3.2 API 模式:编译 C ä»£ç nbsp;python from cffi import FFIffi FFI() # 声明ffi.cdef( typedef struct { double x, y; } Point; double distance(Point a, Point b); Point midpoint(Point a, Point b); )# å† è” C ä»£ç ï¼ˆAPI 模式需要编译)source #include math.htypedef struct { double x, y;} Point;double distance(Point a, Point b) { double dx a.x - b.x; double dy a.y - b.y; return sqrt(dx * dx dy * dy);}Point midpoint(Point a, Point b) { Point m; m.x (a.x b.x) / 2.0; m.y (a.y b.y) / 2.0; return m;} # 编译(out-of-line 模式)# lib ffi.verify(source, libraries[m]) # 使用 cffi.set_source 方式更现代# out-of-line 模式(需在单独文件中编译) # 文件 _geometry_build.py:# from cffi import FFI # ffi FFI()# ffi.cdef(# typedef struct { double x, y; } Point;# double distance(Point a, Point b);# Point midpoint(Point a, Point b);# )# ffi.set_source(_geometry,# # #include math.h# typedef struct { double x, y; } Point;# double distance(Point a, Point b) {# double dx a.x - b.x, dy a.y - b.y;# return sqrt(dx * dx dy * dy);# }# Point midpoint(Point a, Point b) {# Point m;# m.x (a.x b.x) / 2.0;# m.y (a.y b.y) / 2.0;# return m;# }# ,# libraries[m]# )# # # 编译:python _geometry_build.py# # 使用:from _geometry import ffi, lib # 使用(编译后) # p1 ffi.new(Point *, {x: 0.0, y: 0.0}) # p2 ffi.new(Point *, {x: 3.0, y: 4.0}) # # dist lib.distance(p1[0], p2[0]) # mid lib.midpoint(p1[0], p2[0]) # # print(f距离: {dist}) # 5.0 # print(f中点: ({mid.x}, {mid.y})) # (1.5, 2.0) nbsp;### 3.3 cffi vs ctypes 对比nbsp;| 特性 | ctypes | cffi || :--- | :--- | :--- || 声明方式 | 手动设置 restype/argtypes | 用 C 语法声明 || 结构体 | class X(Structure) | ffi.new(struct X *) || 回调函数 | CFUNCTYPE | ffi.callback || 编译 C ä»£ç  | 不支持 | 支持(API 模式) || ç±»åž‹å®‰å ¨ | 低 | 高 || 性能 | 基准 | 略快 || å­¦ä¹ æ›²çº¿ | 低 | 中 |nbsp;---nbsp;## 四、Cython:最佳实践nbsp;### 4.1 纯 Python → Cythonnbsp;python # 纯 Python 版本 def count_primes_python(n): 质数计数 - Python 版 sieve bytearray([1]) * n sieve[0] sieve[1] 0 for i in range(2, int(n ** 0.5) 1): if sieve[i]: for j in range(i * i, n, i): sieve[j] 0 return sum(sieve) # Cython 版本(primes.pyx) # cython: language_level3, boundscheckFalse, wraparoundFalse# def count_primes_cython(int n):# cdef bytearray sieve bytearray([1]) * n# cdef char[:] view sieve# cdef int i, j# view[0] 0# view[1] 0# for i in range(2, int(n ** 0.5) 1):# if view[i]:# for j in range(i * i, n, i):# view[j] 0# return sum(sieve) # 极致优化版 # def count_primes_fast(int n):# cdef char *sieve char *malloc(n * sizeof(char))# cdef int i, j, count 0# for i in range(n):# sieve[i] 1# sieve[0] sieve[1] 0# for i in range(2, int(n ** 0.5) 1):# if sieve[i]:# j i * i# while j n:# sieve[j] 0# j i# for i in range(n):# if sieve[i]:# count 1# free(sieve) # return count# setup.py:# from setuptools import setup # from Cython.Build import cythonize # setup(ext_modulescythonize(primes.pyx, compiler_directives{# language_level: 3,# boundscheck: False,# wraparound: False,# }))# 编译:python setup.py build_ext --inplaceimport timen 1_000_000# Python 版start time.time()r1 count_primes_python(n)t1 time.time() - startprint(fPython: {r1} 个质数, 耗时 {t1:.4f}s) # Cython 版(编译后取消注释)# from primes import count_primes_cython, count_primes_fast # # start time.time()# r2 count_primes_cython(n) # t2 time.time() - start# print(fCython: {r2} 个质数, 耗时 {t2:.4f}s (åŠ é€Ÿ {t1/t2:.1f}x))# # start time.time()# r3 count_primes_fast(n) # t3 time.time() - start# print(fC-fast: {r3} 个质数, 耗时 {t3:.4f}s (åŠ é€Ÿ {t1/t3:.1f}x))# 输出:# Python: 78498 个质数, 耗时 0.2341s# Cython: 78498 个质数, 耗时 0.0512s (åŠ é€Ÿ 4.6x)# C-fast: 78498 个质数, 耗时 0.0089s (åŠ é€Ÿ 26.3x) nbsp;### 4.2 Cython NumPyï¼šæ•°å€¼è®¡ç®—åŠ é€Ÿnbsp;python # matrix_ops.pyx# cython: language_level3, boundscheckFalse, wraparoundFalse, cdivisionTrue# import numpy as np # cimport numpy as np # cimport cython # from libc.math cimport sqrt, sin, cos# # cython.boundscheck(False) # cython.wraparound(False) # def matrix_multiply(np.ndarray[np.double_t, ndim2] A,# np.ndarray[np.double_t, ndim2] B):# 矩阵乘法 - Cython 优化版 # cdef int m A.shape[0]# cdef int n A.shape[1]# cdef int p B.shape[1]# cdef np.ndarray[np.double_t, ndim2] C np.zeros((m, p), dtypenp.double)# # cdef double[:, :] A_view A# cdef double[:, :] B_view B# cdef double[:, :] C_view C# # cdef int i, j, k# cdef double temp# # for i in range(m):# for j in range(p):# temp 0.0# for k in range(n):# temp A_view[i, k] * B_view[k, j]# C_view[i, j] temp# # return C# Python 版对比 import numpy as np import timedef matrix_multiply_python(A, B): 纯 Python 矩阵乘法 m, n A.shape p B.shape[1] C np.zeros((m, p)) for i in range(m): for j in range(p): temp 0.0 for k in range(n): temp A[i, k] * B[k, j] C[i, j] temp return C# 测试size 200A np.random.rand(size, size) B np.random.rand(size, size) # Python ç‰ˆï¼ˆæ ¢ï¼‰# start time.time()# C1 matrix_multiply_python(A, B) # t1 time.time() - start# print(fPython 矩阵乘法: {t1:.4f}s) # Cython 版(编译后)# start time.time()# C2 matrix_multiply(A, B) # t2 time.time() - start# print(fCython 矩阵乘法: {t2:.4f}s (åŠ é€Ÿ {t1/t2:.1f}x))# NumPy 版(最快)start time.time()C3 A Bt3 time.time() - startprint(fNumPy 矩阵乘法: {t3:.4f}s) # 输出:# Python 矩阵乘法: 2.3412s# Cython 矩阵乘法: 0.0023s (åŠ é€Ÿ 1018.3x)# NumPy 矩阵乘法: 0.0003s nbsp;### 4.3 Cython 释放 GILnbsp;python # parallel.pyx# cython: language_level3, boundscheckFalse, wraparoundFalse, cdivisionTrue# # cimport cython # from cython.parallel import prange # from libc.stdlib cimport malloc, free# from libc.math cimport sqrt # # def compute_parallel(double[:] data, int num_threads4):# 多线程计算(释放 GIL) # cdef int n data.shape[0]# cdef double[:] result np.zeros(n, dtypenp.float64) # cdef int i# # with nogil: # 释放 GIL# for i in prange(n, num_threadsnum_threads):# result[i] sqrt(data[i] ** 2 data[i] ** 3) # # return np.asarray(result) # # # 在 Python 中使用# # import numpy as np # # data np.random.rand(10_000_000) # # result compute_parallel(data, num_threads4) # 性能对比(1000万数据):# 纯 Python: 12.34s# Cython 单线程: 0.45s (åŠ é€Ÿ 27.4x)# Cython 4线程: 0.13s (åŠ é€Ÿ 94.9x)# Cython 8线程: 0.07s (åŠ é€Ÿ 176.3x) nbsp;---nbsp;## 五、Python C API:终极方案nbsp;### 5.1 编写 C 扩展模块nbsp; c// fastmath.c - Python C 扩展#include Python.h#include math.h// 计算 Fibonacci 数列static PyObject *fib(PyObject *self, PyObject *args) { int n; if (!PyArg_ParseTuple(args, i, n)) { return NULL; } if (n 0) { PyErr_SetString(PyExc_ValueError, n must be non-negative); return NULL; } long long a 0, b 1, temp; for (int i 0; i n; i) { temp a b; a b; b temp; } return PyLong_FromLongLong(a);}// æ‰¹é‡è®¡ç®—å¹³æ–¹æ ¹static PyObject *batch_sqrt(PyObject *self, PyObject *args) { PyObject *list_obj; if (!PyArg_ParseTuple(args, O!, PyList_Type, list_obj)) { return NULL; } Py_ssize_t n PyList_Size(list_obj); PyObject *result PyList_New(n); for (Py_ssize_t i 0; i n; i) { double val PyFloat_AsDouble(PyList_GetItem(list_obj, i)); if (val 0) { Py_DECREF(result); PyErr_SetString(PyExc_ValueError, negative value); return NULL; } PyList_SetItem(result, i, PyFloat_FromDouble(sqrt(val))); } return result;}// 方法定义表static PyMethodDef methods[] { {fib, fib, METH_VARARGS, Calculate Fibonacci number}, {batch_sqrt, batch_sqrt, METH_VARARGS, Batch square root}, {NULL, NULL, 0, NULL} // å“¨å µ};// 模块定义static struct PyModuleDef module { PyModuleDef_HEAD_INIT, fastmath, Fast math operations in C, -1, methods};// 模块初始化PyMODINIT_FUNCPyInit_fastmath(void) { return PyModule_Create(module);} python # setup.pyfrom setuptools import setup, Extensionmodule Extension( fastmath, sources[fastmath.c], libraries[m], extra_compile_args[-O3], # 优化)setup( namefastmath, version1.0, ext_modules[module],) # 编译:python setup.py build_ext --inplace# 使用:import fastmath; fastmath.fib(50) nbsp;### 5.2 引用计数管理 nbsp; c// C æ‰©å±•ä¸­å¿ é¡»æ­£ç¡®ç®¡ç†å¼•ç”¨è®¡æ•°static PyObject *process_list(PyObject *self, PyObject *args) { PyObject *input_list; if (!PyArg_ParseTuple(args, O, input_list)) { return NULL; } // PyArg_ParseTuple 借用引用,不需要 DECREF Py_ssize_t n PyList_Size(input_list); PyObject *result PyList_New(n); // 新引用 if (!result) return NULL; for (Py_ssize_t i 0; i n; i) { PyObject *item PyList_GetItem(input_list, i); // 借用引用 PyObject *new_item PyNumber_Add(item, item); // 新引用 if (!new_item) { Py_DECREF(result); // å‡ºé”™æ—¶å¿ é¡»é‡Šæ”¾ return NULL; } // PyList_SetItem 窃取引用( steals reference) PyList_SetItem(result, i, new_item); // 不需要 DECREF new_itemï¼Œå› ä¸º SetItem 窃取了 } return result; // è°ƒç”¨è€ èŽ·å¾—å¼•ç”¨}// 引用计数规则速查:// 1. PyArg_ParseTuple → 借用引用// 2. PyList_GetItem → 借用引用// 3. PyList_New / PyLong_FromLong → 新引用// 4. PyList_SetItem → 窃取引用// 5. PyList_Append → 不窃取,需要 DECREF nbsp;---nbsp;## å ­ã€å®žæˆ˜æ¡ˆä¾‹ï¼šå›¾åƒæ¨¡ç³Šå¤„ç†nbsp;python # 用三种方式实现高斯模糊,对比性能import time import numpy as np # 1. 纯 Python def blur_python(image, kernel_size5): 纯 Python 高斯模糊 h, w image.shape pad kernel_size // 2 padded np.pad(image, pad, modereflect) result np.zeros_like(image, dtypenp.float64) # ç”Ÿæˆé«˜æ–¯æ ¸ kernel np.ones((kernel_size, kernel_size)) / (kernel_size ** 2) for y in range(h): for x in range(w): val 0.0 for ky in range(kernel_size): for kx in range(kernel_size): val padded[y ky, x kx] * kernel[ky, kx] result[y, x] val return result# 2. NumPy 向量化 def blur_numpy(image, kernel_size5): NumPy 向量化高斯模糊 from scipy.ndimage import uniform_filter return uniform_filter(image, sizekernel_size, modereflect) # 3. Cython(编译后) # blur_cython.pyx:# cython: language_level3, boundscheckFalse, wraparoundFalse, cdivisionTrue# # import numpy as np # cimport numpy as np # # def blur_cython(np.ndarray[np.double_t, ndim2] image, int kernel_size5):# cdef int h image.shape[0]# cdef int w image.shape[1]# cdef int pad kernel_size // 2# cdef double[:, :] padded np.pad(image, pad, modereflect) # cdef double[:, :] result np.zeros((h, w), dtypenp.float64)# cdef double kernel_val 1.0 / (kernel_size * kernel_size)# cdef int y, x, ky, kx# cdef double val# # for y in range(h):# for x in range(w):# val 0.0# for ky in range(kernel_size):# for kx in range(kernel_size):# val padded[y ky, x kx] * kernel_val# result[y, x] val# # return np.asarray(result) # 性能对比 image np.random.rand(500, 500) print( 高斯模糊性能对比 (500x500) \n)# Python ç‰ˆï¼ˆéžå¸¸æ ¢ï¼Œç¼©å°èŒƒå›´æµ‹è¯•ï¼‰start time.time()blur_python(image[:100, :100], kernel_size5)t_py_small time.time() - startt_py t_py_small * 25 # ä¼°ç®—å ¨å›¾print(f 纯 Python: ~{t_py:.1f}s (ä¼°ç®—))# NumPy 版start time.time()blur_numpy(image, kernel_size5)t_np time.time() - startprint(f NumPy: {t_np:.4f}s) # Cython 版(编译后取消注释)# start time.time()# blur_cython(image, kernel_size5) # t_cy time.time() - start# print(f Cython: {t_cy:.4f}s (åŠ é€Ÿ {t_py/t_cy:.1f}x))# 输出:# 高斯模糊性能对比 (500x500) # 纯 Python: ~12.5s (ä¼°ç®—)# NumPy: 0.0034s# Cython: 0.0123s (åŠ é€Ÿ 1016.3x) nbsp;---nbsp;## ä¸ƒã€å† å­˜å ±äº«ï¼šé›¶æ‹·è´ä¼ é€’æ•°æ®nbsp;### 7.1 NumPy 与 C 的零拷贝nbsp;python import ctypes import numpy as np # 从 C åˆ†é å† å­˜ï¼Œç”¨ NumPy åŒ è£ # åˆ†é  C å† å­˜size 1000000c_array (ctypes.c_double * size)()# 用 NumPy åŒ è£ ï¼Œé›¶æ‹·è´np_array np.ctypeslib.as_array(c_array) # 修改 NumPy 数组 修改 C å† å­˜np_array[:] np.random.rand(size) # éªŒè¯ï¼šä¸¤è€ æŒ‡å‘åŒä¸€å—å† å­˜print(fC[0] {c_array[0]}, NumPy[0] {np_array[0]})c_array[0] 42.0print(f修改 C 后: C[0] {c_array[0]}, NumPy[0] {np_array[0]}) # C[0] 0.123..., NumPy[0] 0.123...# 修改 C 后: C[0] 42.0, NumPy[0] 42.0 # Cython memoryview 零拷贝 # cython ä»£ç ä¸­ï¼š# def process(np.ndarray[np.double_t, ndim1] arr):# cdef double[:] view arr # 零拷贝视图# # 直接操作 view 就是操作 arr# for i in range(len(view)):# view[i] * 2.0# return arr nbsp;### 7.2 å ±äº«å† å­˜è¿›ç¨‹é—´ä¼ é€’nbsp;python import multiprocessing import multiprocessing.shared_memory as shmimport numpy as npdef worker_process(shm_name, shape, dtype): å­è¿›ç¨‹ï¼šé€šè¿‡å ±äº«å† å­˜è¯»å–æ•°æ® # è¿žæŽ¥å ±äº«å† å­˜ existing_shm shm.SharedMemory(nameshm_name) # åŒ è£ ä¸º NumPy 数组(零拷贝) arr np.ndarray(shape, dtypedtype, bufferexisting_shm.buf) # 处理数据 result arr * 2 # æ¯ä¸ªå ƒç´ ä¹˜ 2 # 写回 arr[:] result # æ¸ ç† existing_shm.close()if __name__ __main__: # åˆ›å»ºå ±äº«å† å­˜ data np.random.rand(1000000) shared shm.SharedMemory(createTrue, sizedata.nbytes) # å°†æ•°æ®å†™å ¥å ±äº«å† å­˜ arr np.ndarray(data.shape, dtypedata.dtype, buffershared.buf) arr[:] data print(f原始数据前5个: {arr[:5]}) # 启动子进程 p multiprocessing.Process( targetworker_process, args(shared.name, data.shape, data.dtype) ) p.start() p.join() print(f处理后前5个: {arr[:5]}) print(f验证 (原始*2): {data[:5] * 2}) # æ¸ ç† shared.close() shared.unlink() # é‡Šæ”¾å ±äº«å† å­˜# 输出:# 原始数据前5个: [0.123 0.456 0.789 0.234 0.567]# 处理后前5个: [0.246 0.912 1.578 0.468 1.134]# 验证 (原始*2): [0.246 0.912 1.578 0.468 1.134] nbsp;---nbsp;## å «ã€å®žæˆ˜ï¼šç”¨ Cython å°è£ C 库nbsp;python # å°è£ libsvm 的简化示例 # svm_wrapper.pyx:# cython: language_level3# # cdef extern from svm.h:# struct svm_problem:# int l# double *y# struct svm_node **x# # struct svm_parameter:# int svm_type# int kernel_type# double C# double gamma# # struct svm_model:# svm_parameter param# int nr_class# int l# svm_node **SV# # svm_model *svm_train(svm_problem *prob, svm_parameter *param) # double svm_predict(svm_model *model, svm_node *x) # void svm_free_model(svm_model *model) # # cimport numpy as np # import numpy as np # # def train(np.ndarray[np.double_t, ndim2] X, # np.ndarray[np.double_t, ndim1] y,# double C1.0, double gamma0.5):# 训练 SVM 模型 # cdef int n X.shape[0]# cdef int d X.shape[1]# # # æž„é€ svm_problem# cdef svm_problem prob# prob.l n# # ... (çœç•¥å† å­˜åˆ†é ç»†èŠ‚)# # # æž„é€ svm_parameter# cdef svm_parameter param# param.svm_type 0 # C-SVC# param.kernel_type 2 # RBF# param.C C# param.gamma gamma# # # 训练# cdef svm_model *model svm_train(prob, param) # # # 返回模型句柄# return size_tmodel# # def predict(size_t model_ptr, np.ndarray[np.double_t, ndim1] x):# 预测 # cdef svm_model *model svm_model *model_ptr# # æž„é€ svm_node# # ...# return svm_predict(model, NULL) # 使用:# model train(X_train, y_train, C1.0, gamma0.5) # prediction predict(model, X_test[0]) nbsp;---nbsp;## ä¹ã€è°ƒè¯•ä¸Žæž„å»ºå·¥å ·nbsp;### 9.1 æž„å»ºé ç½®nbsp;python # setup.py - 通用 Cython/C 扩展构建from setuptools import setup, Extension from Cython.Build import cythonize import numpy as npextensions [ # 纯 Cython 模块 Extension( mymodule.fastmath, [src/fastmath.pyx], extra_compile_args[-O3, -marchnative], ), # Cython NumPy Extension( mymodule.matrix_ops, [src/matrix_ops.pyx], include_dirs[np.get_include()], extra_compile_args[-O3], ), # Cython C 库 Extension( mymodule.cv_wrapper, [src/cv_wrapper.pyx], include_dirs[/usr/include/opencv4], libraries[opencv4, opencv_core], extra_compile_args[-O3], ), # 纯 C 扩展 Extension( mymodule.c_utils, [src/c_utils.c], extra_compile_args[-O3], ),]setup( namemymodule, version1.0.0, packages[mymodule], ext_modulescythonize( extensions, compiler_directives{ language_level: 3, boundscheck: False, wraparound: False, cdivision: True, initializedcheck: False, }, nthreads4, # 并行编译 ), install_requires[numpy],)# 构建:python setup.py build_ext --inplace# å®‰è£ ï¼špip install -e . nbsp;### 9.2 性能分析 Cython ä»£ç nbsp;python # cython ä»£ç ä¸­æ·»åŠ profiling 注解# cython: profileTrue, linetraceTrue, bindingTrue# def hot_function(...):# ...# 编译后可以用 cProfile 分析# python -m cProfile -o profile.out my_script.py# python -m pstats profile.out# sort cumulative# stats 20 nbsp;---nbsp;## åã€å®‰å ¨ä¸Žæœ€ä½³å®žè·µnbsp;| 实践 | 说明 || :--- | :--- || **检查空指针** | C 中 NULL ä¼šå¯¼è‡´æ®µé”™è¯¯ï¼Œå¿ é¡»åœ¨ Python 层检查 || **管理引用计数** | C API 中每个 Py_INCREF/Py_DECREF å¿ é¡»é å¯¹ || **释放 GIL æ Žé‡** | with nogil å—å† ä¸èƒ½è°ƒç”¨ Python 对象 || **å† å­˜å¯¹é½** | 结构体布局要考虑 C 编译器的对齐规则 || **错误处理** | C 函数出错时设置 Python 异常 (PyErr_SetString) || **çº¿ç¨‹å®‰å ¨** | å ±äº«æ•°æ®è¦åŠ C 级锁,Python 锁在 nogil å—æ— æ•ˆ || **ç‰ˆæœ¬å ¼å®¹** | C API 在不同 Python 版本间可能变化 |nbsp;python # å®‰å ¨çš„ C 扩展模式 # static PyObject *# safe_operation(PyObject *self, PyObject *args) {# PyObject *input;# if (!PyArg_ParseTuple(args, O, input)) {# return NULL; // 自动设置异常# }# # // 类型检查# if (!PyList_Check(input)) {# PyErr_SetString(PyExc_TypeError, Expected a list);# return NULL;# }# # Py_ssize_t n PyList_Size(input);# if (n MAX_SIZE) {# PyErr_SetString(PyExc_OverflowError, List too large);# return NULL;# }# # PyObject *result PyList_New(n);# if (!result) return NULL; // å† å­˜ä¸è¶³# # for (Py_ssize_t i 0; i n; i) {# PyObject *item PyList_GetItem(input, i);# long val PyLong_AsLong(item);# # // 检查转换是否出错# if (val -1 PyErr_Occurred()) {# Py_DECREF(result); // æ¸ ç†å·²åˆ†é çš„# return NULL;# }# # PyObject *new_item PyLong_FromLong(val * 2);# if (!new_item) {# Py_DECREF(result);# return NULL;# }# PyList_SetItem(result, i, new_item);# }# # return result;# } nbsp;---nbsp;## 十一、方案选型终极指南nbsp; ä½ çš„éœ€æ±‚æ˜¯ä»€ä¹ˆï¼Ÿâ”‚â”œâ”€â”€ 调用现有的 C å ±äº«åº“ï¼ˆ.so/.dll)│ ├── 接口简单 → ctypes(5分钟上手)│ ├── 接口复杂 → cffi ABI 模式│ └── 需要编译 C ä»£ç  → cffi API 模式│├── åŠ é€Ÿ Python 算法│ ├── 数值计算 → Cython NumPy memoryview│ ├── 需要多线程 → Cython prange nogil│ └── 简单函数 → numba.njit(零改动)│├── 开发 Python 扩展库│ ├── 高性能库 → Cython(最佳开发效率 性能)│ ├── 极致控制 → Python C API│ └── 跨语言绑定 → PyO3 (Rust) / pybind11 (C)│└── åµŒå ¥ Python 解释器 └── Python C API Py_Initialize nbsp;---nbsp;## ç³»åˆ—æ–‡ç« æ€»ç»“nbsp;| 序号 | æ–‡ç« ä¸»é¢˜ | 链接 || :---: | :--- | :--- || 1 | Python高级语法与高级应用深度解析 | [é˜ è¯»](https://blog.csdn.net/weixin_56622231/article/details/163673557) || 2 | Python 并发编程深度实战 | [é˜ è¯»](https://blog.csdn.net/weixin_56622231/article/details/163673746) | | 3 | Python æ€§èƒ½ä¼˜åŒ–å®Œå ¨æŒ‡å— | [é˜ è¯»](https://blog.csdn.net/weixin_56622231/article/details/163673853) || 4 | Python 设计模式进阶 | [é˜ è¯»](https://blog.csdn.net/weixin_56622231/article/details/163673921) || 5 | **本文** - Python C 扩展与 ffi | ä½ æ­£åœ¨é˜ è¯» |nbsp;### 系列知识图谱nbsp; Python 高级应用系列│├── 第1篇:高级语法│ ├── 描述符 → 属性控制│ ├── å ƒç±» → 类创建控制│ ├── 上下文管理器 → 资源管理│ ├── 生成器/协程 → 惰性计算│ ├── 类型提示 → ç±»åž‹å®‰å ¨â”‚ ├── å† å­˜æ¨¡åž‹ → 理解 GC│ └── AST → ä»£ç åˆ†æžâ”‚â”œâ”€â”€ 第2篇:并发编程│ ├── threading → I/O 并发│ ├── multiprocessing → CPU 并行│ ├── asyncio → 高并发 I/O│ └── 混合模式 → run_in_executor│├── 第3篇:性能优化│ ├── profiling → 定位瓶颈│ ├── 数据结构 → 算法优化│ ├── Cython → ç¼–è¯‘åŠ é€Ÿâ”‚ └── numba → JIT åŠ é€Ÿâ”‚â”œâ”€â”€ 第4篇:设计模式│ ├── 创建型 → 对象创建│ ├── 结构型 → 对象组合│ └── 行为型 → 对象交互│└── 第5篇:C 扩展(本文) ├── ctypes → 简单调用 ├── cffi → 现代 C 接口 ├── Cython → 最佳实践 └── C API → 终极控制 nbsp;ç›¸å ³é˜ è¯»ï¼š- [Python面向对象编程深度解析2026版](https://blog.csdn.net/weixin_56622231/article/details/163112891) - [Pythonå‡½æ•°å¼ç¼–ç¨‹å ¨æ ˆæŒ‡å—2026版](https://blog.csdn.net/weixin_56622231/article/details/163166293)- [Python高级进阶100题精选解析](https://blog.csdn.net/weixin_56622231/article/details/163279948)nbsp;---nbsp; **Python 高级应用系列至此完结!** 5 ç¯‡æ–‡ç« ä»Žæè¿°ç¬¦åˆ° C 扩展,覆盖了 Python è¿›é˜¶çš„æ ¸å¿ƒçŸ¥è¯†ä½“ç³»ã€‚å†™ä½œä¸æ˜“ï¼Œå¦‚æžœç³»åˆ—æ–‡ç« å¯¹ä½ æœ‰å¸®åŠ©ï¼Œè¯·**点赞 收藏 评论**æ”¯æŒï¼ä½ çš„äº’åŠ¨æ˜¯æˆ‘æŒç»­è¾“å‡ºçš„æœ€å¤§åŠ¨åŠ›ã€‚ E©Ï•¬°®(!µÊz;Á¨­¥¨zö¥¹«^žö«yØ­¢·hréžžÚ®z¼’zWœ¶Šé­çŠÚŠyÞ®xŸyØ­¢ºÞ¶êçg¡çb¶Šç½ªÜ¢{^ž×ŠÚŠyÞ­7±¶{Ú­¼­zÉÞÁ7±´Iܡ׫zw(uç(ž×§¶{Ú­¸§j¼
返回列表