
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¼