Python 性能优化完全指南2026版：从 profiling 到 Cython 的全链路加速实战

Python 性能优化完全指南2026版从 profiling 到 Cython 的全链路加速实战本文是 Python 高级应用系列的第 3 篇。建议配合阅读 第 1 篇高级语法 和 第 2 篇并发编程。 — ## 一、性能优化的正确心态 “过早优化是万恶之源。”—— Donald Knuth 性能优化不是盲目改代码而是一个测量 → 分析 → 优化 → 验证的闭环 发现问题 → 测量基线 → 定位瓶颈 → 选择策略 → 实施优化 → 验证效果 → 回归测试 ↑ | └──────────────────────── 不达标则循环 ←──────────────────────────────┘| 优化层级 | 手段 | 提升幅度 | 难度 || :—: | :— | :—: | :—: ||算法层| 换更好的算法/数据结构 | 10x ~ 1000x | 中 ||架构层| 并发/并行/缓存/批处理 | 5x ~ 50x | 中高 ||语言层| 内置函数/列表推导/生成器 | 2x ~ 10x | 低 ||编译层| Cython/numba/PyPy | 10x ~ 100x | 中 ||系统层| 内存映射/零拷贝/IO优化 | 2x ~ 10x | 高 | — ## 二、性能测量工具箱 ### 2.1 timeit微基准测试 pythonimport timeit 基本用法 # 测量列表拼接的两种方式t1 timeit.timeit(“‘’.join([‘a’, ‘b’, ‘c’, ‘d’])”, number100000)t2 timeit.timeit(“‘a’ ‘b’ ‘c’ ‘d’”, number100000)print(fjoin: {t1:.4f}s | : {t2:.4f}s | join 快 {t2/t1:.1f}x)# 输出join: 0.0152s | : 0.0231s | join 快 1.5x# 命令行用法 # python -m timeit -s “x list(range(1000))” “sum(x)”# python -m timeit -s “x list(range(1000))” “sum(x)” -n 10000 -r 5# 多方案对比 def benchmark_dict(): “”“字典 vs 列表查找性能”“”setup “”data_list list(range(10000))data_dict {i: i for i in range(10000)} “”t_list timeit.timeit(“9999 in data_list”, setupsetup, number10000) t_dict timeit.timeit(“9999 in data_dict”, setupsetup, number10000) print(f 列表查找: {t_list:.4f}s) print(f 字典查找: {t_dict:.4f}s) print(f 字典快: {t_list/t_dict:.1f}x)benchmark_dict()# 输出# 列表查找: 2.8431s# 字典查找: 0.0003s# 字典快: 9476.9x### 2.2 cProfile函数级性能分析 pythonimport cProfileimport pstatsimport iodef slow_function(): “”“模拟慢函数”“”total 0 for i in range(1000000):total i return totaldef fast_function(): “”模拟快函数“”return sum(range(1000000))def main(): for _ in range(10):slow_function() for _ in range(10):fast_function()# 基本用法 cProfile.run(‘main()’, sort‘cumulative’)# 输出示例# 32 function calls in 0.891 seconds# Ordered by: cumulative time# ncalls tottime percall cumtime percall filename:lineno(function)1 0.000 0.000 0.891 0.891 :1()1 0.000 0.000 0.891 0.891 test.py:9(main)10 0.762 0.076 0.762 0.076 test.py:2(slow_function)10 0.129 0.013 0.129 0.013 test.py:7(fast_function) 更精细的控制 def profile_code(func, *args, **kwargs): “”性能分析装饰器“”pr cProfile.Profile() pr.enable() result func(*args, **kwargs) pr.disable() s io.StringIO() ps pstats.Stats(pr, streams).sort_stats(‘cumulative’) ps.print_stats(10) # 前 10 个最耗时的函数 print(s.getvalue()) return result# 使用profile_code(main)### 2.3 line_profiler逐行性能分析 python安装pip install line_profilerfrom line_profiler import LineProfilerdef process_data(data): “”数据处理函数“”result [] for item in data:过滤 if item % 2 0:# 变换 value item ** 2 result.append(value) # 排序 result.sort() # 聚合 total sum(result) avg total / len(result) if result else 0 return total, avg# 逐行分析lp LineProfiler()lp_wrapper lp(process_data)lp_wrapper(list(range(100000)))lp.print_stats()# 输出示例# Timer unit: 1e-06 s# Total time: 0.045678 s# File: test.py# Function: process_data at line 3## Line # Hits Time Per Hit % Time Line Contents# # 3 def process_data(data):# 4 1 2.0 2.0 0.0 result []# 5 100001 35000.0 0.3 76.6 for item in data:# 6 100000 15000.0 0.1 32.8 if item % 2 0:# 7 50000 8000.0 0.2 17.5 value item ** 2# 8 50000 6000.0 0.1 13.1 result.append(value)9 1 1.0 1.0 0.0 result.sort()# 10 1 500.0 500.0 1.1 total sum(result)11 1 10.0 10.0 0.0 avg total / len(result) if result else 0# 12 1 1.0 1.0 0.0 return total, avg## 可以看到 76.6% 的时间花在 for 循环上### 2.4 memory_profiler内存分析 python安装pip install memory_profilerfrom memory_profiler import profileprofiledef memory_heavy():内存消耗大的函数“”# 创建大列表 data [i * 2 for i in range(1000000)] # 复制 data_copy data[:] # 过滤 filtered [x for x in data if x % 3 0] # 清理 del data del data_copy return len(filtered)memory_heavy()# 输出示例# Line # Mem usage Increment Occurrences Line Contents# # 4 50.1 MiB 50.1 MiB 1 profile# 5 50.1 MiB 0.0 MiB 1 def memory_heavy():# 7 65.4 MiB 15.3 MiB 1 data [i * 2 for i in range(1000000)]# 9 80.7 MiB 15.3 MiB 1 data_copy data[:]# 11 85.9 MiB 5.2 MiB 1 filtered [x for x in data if x % 3 0]# 13 70.6 MiB -15.3 MiB 1 del data# 14 55.3 MiB -15.3 MiB 1 del data_copy# 16 55.3 MiB 0.0 MiB 1 return len(filtered)— ## 三、Python 语言层优化 ### 3.1 数据结构选择 pythonimport timeit 列表 vs 集合成员查找 setup “”data_list list(range(100000))data_set set(range(100000))“”“t1 timeit.timeit(“99999 in data_list”, setupsetup, number10000)t2 timeit.timeit(“99999 in data_set”, setupsetup, number10000)print(f列表查找: {t1:.4f}s | 集合查找: {t2:.4f}s | 集合快 {t1/t2:.0f}x”)# 列表查找: 2.8413s | 集合查找: 0.0003s | 集合快 9471x# 列表 vs deque头部插入 from collections import dequedef list_prepend(n): lst [] for i in range(n):lst.insert(0, i) return lstdef deque_prepend(n): dq deque() for i in range(n):dq.appendleft(i) return list(dq)n 50000t1 timeit.timeit(lambda: list_prepend(n), number10)t2 timeit.timeit(lambda: deque_prepend(n), number10)print(flist头部插入: {t1:.4f}s | deque头部插入: {t2:.4f}s | deque快 {t1/t2:.0f}x)list头部插入: 5.2341s | deque头部插入: 0.0152s | deque快 344x# 字符串拼接 def concat_plus(n): s “” for i in range(n):s str(i) return sdef concat_join(n): return .join(str(i) for i in range(n))n 10000t1 timeit.timeit(lambda: concat_plus(n), number100)t2 timeit.timeit(lambda: concat_join(n), number100)print(f拼接: {t1:.4f}s | join: {t2:.4f}s | join快 {t1/t2:.1f}x)拼接: 0.4521s | join: 0.0312s | join快 14.5x### 3.2 列表推导 vs 循环 pythonimport timeit 列表推导更快 def loop_version(n): result [] for i in range(n):if i % 2 0: result.append(i ** 2) return resultdef comprehension_version(n): return [i ** 2 for i in range(n) if i % 2 0]def filter_map_version(n): return list(map(lambda x: x ** 2, filter(lambda x: x % 2 0, range(n))))n 1000000t1 timeit.timeit(lambda: loop_version(n), number5)t2 timeit.timeit(lambda: comprehension_version(n), number5)t3 timeit.timeit(lambda: filter_map_version(n), number5)print(ffor循环: {t1:.4f}s)print(f列表推导: {t2:.4f}s (快 {t1/t2:.1f}x)“)print(ffiltermap: {t3:.4f}s (快 {t1/t3:.1f}x)”)# 输出# for循环: 0.8234s# 列表推导: 0.6521s (快 1.3x)# filtermap: 0.7102s (快 1.2x)nbsp;### 3.3 生成器节省内存nbsp;pythonimport sys 列表 vs 生成器内存对比 data_list [i ** 2 for i in range(1000000)]data_gen (i ** 2 for i in range(1000000))print(f列表内存: {sys.getsizeof(data_list) / 1024 / 1024:.1f} MB)print(f生成器内存: {sys.getsizeof(data_gen)} bytes)# 列表内存: 8.4 MB# 生成器内存: 200 bytes# 大文件处理生成器流水线 def read_lines(filepath): “”“逐行读取大文件”“”with open(filepath, r) as f: for line in f: yield line.strip()def parse_csv(lines): 解析 CSV 行“”for line in lines:yield line.split(‘,’)def filter_empty(rows): “”过滤空行“”for row in rows:if len(row) 3:yield rowdef transform(rows): “”数据变换“”for row in rows:yield { ‘name’: row[0], ‘age’: int(row[1]), ‘score’: float(row[2]) }# 流水线每一步都是惰性的# pipeline transform(filter_empty(parse_csv(read_lines(‘large_file.csv’))))# for record in pipeline:# process(record)整个过程内存占用恒定不管文件多大### 3.4 内置函数的威力 pythonimport timeitfrom operator import itemgetter, attrgetterdata [(i, i ** 2, i ** 3) for i in range(100000)] 排序key 函数 vs lambda t1 timeit.timeit(lambda: sorted(data, keylambda x: x[1]), number10)t2 timeit.timeit(lambda: sorted(data, keyitemgetter(1)), number10)print(flambda排序: {t1:.4f}s | itemgetter排序: {t2:.4f}s | itemgetter快 {t1/t2:.1f}x)lambda排序: 0.4521s | itemgetter排序: 0.3812s | itemgetter快 1.2x# sum vs 循环 numbers list(range(1000000))t1 timeit.timeit(lambda: sum(numbers), number10)t2 timeit.timeit(lambda: sum(x for x in numbers), number10) # 生成器t3 timeit.timeit(“”total 0for x in numbers:total x, setup“numbers list(range(1000000))”, number10)print(fsum(list): {t1:.4f}s | sum(gen): {t2:.4f}s | for循环: {t3:.4f}s)# sum(list): 0.0452s | sum(gen): 0.1023s | for循环: 0.0651s# 注意sum 直接传列表比传生成器更快| 优化技巧 | 提速倍数 | 说明 || :— | :—: | :— || 用set替代list做查找 | 1000x | O(1) vs O(n) || 用join替代拼字符串 | 10x | 避免反复创建新字符串 || 用列表推导替代 for 循环 | 1.2~1.5x | C 层优化 || 用 deque 做队列操作 | 100x | 头部插入 O(1) || 用 itemgetter 替代 lambda | 1.2x | C 层优化 || 用生成器节省内存 | N/A | 空间换不了时间但省内存 | — ## 四、算法层优化 ### 4.1 缓存functools.cache pythonimport functoolsimport time 递归 Fibonacci无缓存 def fib_no_cache(n):if n 1: return n return fib_no_cache(n - 1) fib_no_cache(n 递归 Fibonacci有缓存 functools.cachedef fib_cached(n): if n 1:return n return fib_cached(n - 1) fib_cached(n对比n 35start time.time()r1 fib_no_cache(n)t1 time.time() - startstart time.time()r2 fib_cached(n)t2 time.time()startprint(f无缓存: fib({n}) {r1}, 耗时 {t1:.4f}s)print(f有缓存: fib({n}) {r2}, 耗时 {t2:.6f}s)print(f加速比: {t1/t2:.0f}x)无缓存: fib(35) 9227465, 耗时 2.8431s# 有缓存: fib(35) 9227465, 耗时 0.000012s# 加速比: 236917x# lru_cache限制缓存大小 functools.lru_cache(maxsize128)def fetch_user(user_id): “”“模拟数据库查询”“”time.sleep(0.1) # 模拟查询耗时 return {“id”: user_id, “name”: fUser-{user_id}“}# 第一次查询慢start time.time()fetch_user(1)print(f第一次查询: {time.time() - start:.4f}s”) # 0.1023s# 第二次查询命中缓存start time.time()fetch_user(1)print(f第二次查询: {time.time() - start:.6f}s) # 0.000002s# 查看缓存信息print(f缓存信息: {fetch_user.cache_info()})# 缓存信息: CacheInfo(hits1, misses1, maxsize128, currsize1)### 4.2 数据结构优化实战 python 场景统计文本词频 import timefrom collections import Counter, defaultdicttext the quick brown fox jumps over the lazy dog * 100000words text.split()方法1普通字典def count_dict(words): freq {} for word in words:if word in freq: freq[word] 1 else: freq[word] 1 return freq# 方法2defaultdictdef count_defaultdict(words): freq defaultdict(int) for word in words: freq[word] 1 return freq# 方法3Counterdef count_counter(words): return Counter(words)方法4dict.getdef count_get(words): freq {} for word in words:freq[word] freq.get(word, 0) 1 return freq# 基准测试for func in [count_dict, count_defaultdict, count_counter, count_get]: start time.time() result func(words) elapsed time.time() - start print(f {func.__name__:25s}: {elapsed:.4f}s)输出# count_dict : 0.0823s# count_defaultdict : 0.0612s# count_get : 0.0734s# count_counter : 0.0321s ← 最快### 4.3 算法复杂度对比 pythonimport timeimport bisect 场景在有序序列中查找 data list(range(1000000))# 线性查找 O(n)def linear_search(arr, target): for i, v in enumerate(arr):if v target:return i return -1# 二分查找 O(log n)def binary_search(arr, target): lo, hi 0, len(arr) - 1 while lo hi: mid (lo hi) // 2 if arr[mid] target:return mid elif arr[mid] target:lo mid 1 else: hi mid - 1 return -1# bisect 模块C 实现def bisect_search(arr, target): idx bisect.bisect_left(arr, target) if idx len(arr) and arr[idx] target: return idx return -1target 999999start time.time()linear_search(data, target)t1 time.time() - startstart time.time()binary_search(data, target)t2 time.time()startstart time.time()bisect_search(data, target)t3 time.time() - startprint(f线性查找: {t1:.6f}s)print(f二分查找: {t2:.6f}s (快 {t1/t2:.0f}x)“)print(fbisect: {t3:.6f}s (快 {t1/t3:.0f}x)”)# 线性查找: 0.034521s# 二分查找: 0.000012s (快 2877x)# bisect: 0.000002s (快 17260x)— ## 五、CythonC 级速度 ### 5.1 Cython 基础 python安装pip install cython# 纯 Python 版本 def primes_python(n): “”“质数筛 - Python 版”“”sieve [True] * n sieve[0] sieve[1] False for i in range(2, int(n ** 0.5) 1): if sieve[i]: for j in range(i * i, n, i): sieve[j] False return [i for i in range(n) if sieve[i]]# Cython 版本保存为 primes.pyx # primes.pyx:# def primes_cython(int n):# cdef list sieve [True] * n# cdef int i, j# sieve[0] sieve[1] False# for i in range(2, int(n ** 0.5) 1):# if sieve[i]:# for j in range(i * i, n, i):# sieve[j] False# return [i for i in range(n) if sieve[i]]# setup.py:# from setuptools import setupfrom Cython.Build import cythonizesetup(ext_modulescythonize(“primes.pyx”))# 编译python setup.py build_ext --inplace# 性能对比 import timen 1000000start time.time()r1 primes_python(n)t1 time.time() - startprint(fPython: {t1:.4f}s, 找到 {len(r1)} 个质数)# from primes import primes_cythonstart time.time()# r2 primes_cython(n)t2 time.time() - start# print(fCython: {t2:.4f}s, 找到 {len(r2)} 个质数)# print(f加速比: {t1/t2:.1f}x)输出# Python: 0.2341s, 找到 78498 个质数# Cython: 0.0152s, 找到 78498 个质数# 加速比: 15.4x### 5.2 Cython 类型注解 python 纯 Python def compute_python(data): total 0.0 for i in range(len(data)):total data[i] ** 2 data[i] ** 0.5 return total# Cython 优化版compute.pyx # def compute_cython(list data):# cdef double total 0.0# cdef double val# cdef int i# cdef int n len(data)for i in range(n):# val data[i]# total val * val val ** 0.5# return total# Cython 极致优化版用 C 数组 # import numpy as npcimport numpy as npcimport cythoncython.boundscheck(False)cython.wraparound(False)def compute_cython_fast(np.ndarray[np.double_t, ndim1] data):# cdef double total 0.0# cdef double val# cdef int i, n len(data)for i in range(n):# val data[i]# total val * val val ** 0.5# return totalimport timeimport numpy as npdata list(np.random.rand(1000000))data_np np.array(data)start time.time()compute_python(data)t1 time.time() - startprint(fPython: {t1:.4f}s)Cython 编译后取消注释# start time.time()# compute_cython(data)t2 time.time() - start# print(fCython: {t2:.4f}s (快 {t1/t2:.1f}x))## start time.time()# compute_cython_fast(data_np)t3 time.time() - start# print(fCythonC数组: {t3:.4f}s (快 {t1/t3:.1f}x))# 输出# Python: 0.4521s# Cython: 0.0823s (快 5.5x)# CythonC数组: 0.0034s (快 133.0x)— ## 六、numbaJIT 即时编译 python安装pip install numbafrom numba import jit, njit, prangeimport numpy as npimport time 纯 Python def monte_carlo_pi_python(n): “”“蒙特卡洛计算 π - Python 版”“”count 0 for _ in range(n): x np.random.random() y np.random.random() if x * x y * y 1.0: count 1 return 4.0 * count / n# numba JIT njitdef monte_carlo_pi_numba(n): 蒙特卡洛计算 π - numba 版“”count 0 for _ in range(n):x np.random.random() y np.random.random() if x * x y * y 1.0:count 1 return 4.0 * count / n# numba 并行 njit(parallelTrue)def monte_carlo_pi_parallel(n): “”蒙特卡洛计算 π - numba 并行版“”count 0 for _ in prange(n):x np.random.random() y np.random.random() if x * x y * y 1.0:count 1 return 4.0 * count / nn 10_000_000# Pythonstart time.time()pi1 monte_carlo_pi_python(n)t1 time.time() - startprint(fPython: π ≈ {pi1:.6f}, 耗时 {t1:.4f}s)numba第一次运行包含编译时间start time.time()pi2 monte_carlo_pi_numba(n)t2 time.time() - startprint(fnumba: π ≈ {pi2:.6f}, 耗时 {t2:.4f}s (含编译))# numba第二次运行已编译start time.time()pi3 monte_carlo_pi_numba(n)t3 time.time()startprint(fnumba: π ≈ {pi3:.6f}, 耗时 {t3:.4f}s (已编译)“)# numba 并行start time.time()pi4 monte_carlo_pi_parallel(n)t4 time.time() - startprint(fnumba并行:π ≈ {pi4:.6f}, 耗时 {t4:.4f}s”)print(f\n加速比: numba {t1/t3:.1f}x | 并行 {t1/t4:.1f}x)# 输出# Python: π ≈ 3.141592, 耗时 5.2341s# numba: π ≈ 3.141592, 耗时 0.8231s (含编译)# numba: π ≈ 3.141592, 耗时 0.0152s (已编译)# numba并行:π ≈ 3.141592, 耗时 0.0041s## 加速比: numba 344.3x | 并行 1276.6x| 工具 | 适用场景 | 提速 | 优点 | 缺点 || :—: | :— | :—: | :— | :— || Cython | 数值计算、库开发 | 10~150x | 极致性能 | 需编译学习曲线 || numba | 数值计算、科学计算 | 50~350x | 零改动自动 JIT | 首次编译慢 || PyPy | 通用 Python | 3~10x | 无需改代码 | 兼容性问题 || numpy | 数组运算 | 10~100x | 向量化生态好 | 仅数值数组 | — ## 七、I/O 优化### 7.1 文件读写优化 pythonimport time 逐行读取 vs 一次性读取 def read_lines(filepath): “”“逐行读取适合大文件”“”with open(filepath) as f: for line in f: process(line)def read_all(filepath): “”一次性读取适合小文件“”with open(filepath) as f: for line in f.readlines():process(line)def read_iter(filepath): “”迭代器方式“”with open(filepath) as f: yield from f# 批量写入 vs 逐行写入 def write_lines_slow(filepath, lines): “”逐行写入慢“”with open(filepath, ‘w’) as f: for line in lines:f.write(line ‘\n’)def write_lines_fast(filepath, lines): “”批量写入快“”with open(filepath, ‘w’) as f: f.write(‘\n’.join(lines))# 测试lines [fline {i} for i in range(1000000)]filepath /tmp/test_write.txt’start time.time()write_lines_slow(filepath, lines)t1 time.time() - startstart time.time()write_lines_fast(filepath, lines)t2 time.time()startprint(f逐行写入: {t1:.4f}s | 批量写入: {t2:.4f}s | 批量快 {t1/t2:.1f}x)逐行写入: 0.2341s | 批量写入: 0.0512s | 批量快 4.6x### 7.2 内存映射大文件 pythonimport mmapimport time 传统方式读取大文件 def traditional_read(filepath): with open(filepath, ‘r’) as f: data f.read() return data# mmap 方式 def mmap_read(filepath): with open(filepath, ‘r’) as f: # 内存映射 mm mmap.mmap(f.fileno(), 0, accessmmap.ACCESS_READ) data mm.read() mm.close() return data# 创建大文件import osfilepath /tmp/large_file.txt’with open(filepath, ‘w’) as f: f.write(‘x’ * 100_000_000) # 100MB# 对比start time.time()traditional_read(filepath)t1 time.time() - startstart time.time()mmap_read(filepath)t2 time.time()startprint(f传统读取: {t1:.4f}s | mmap: {t2:.4f}s | mmap快 {t1/t2:.1f}x)os.remove(filepath)传统读取: 0.1023s | mmap: 0.0312s | mmap快 3.3x— ## 八、缓存策略 ### 8.1 多级缓存 pythonimport functoolsimport timeimport hashlibimport json 内存缓存 class MemoryCache: “”“简易内存缓存”“”def __init__(self, maxsize1000, ttl300): self._cache {} self._maxsize maxsize self._ttl ttl # 过期时间秒 def get(self, key): if key in self._cache: value, timestamp self._cache[key] if time.time() - timestamp self._ttl: return value else: del self._cache[key] # 过期 return None def set(self, key, value): # LRU超过容量时删除最早的 if len(self._cache) self._maxsize: oldest min(self._cache, keylambda k: self._cache[k][1]) del self._cache[oldest] self._cache[key] (value, time.time()) def clear(self): self._cache.clear()cache MemoryCache(maxsize100, ttl60)def cached_api_call(url, paramsNone): “”带缓存的 API 调用“”# 生成缓存 key key_str url json.dumps(params or {}, sort_keysTrue) key hashlib.md5(key_str.encode()).hexdigest() # 查缓存 cached cache.get(key) if cached is not None:print(f [缓存命中] {url}“) return cached # 模拟 API 调用 print(f” [API调用] {url}“) time.sleep(1) # 模拟网络延迟 result {“data”: fresponse from {url}”, “params”: params} # 写缓存 cache.set(key, result) return result# 测试cached_api_call(“https://api.example.com/users”, {“page”: 1})cached_api_call(“https://api.example.com/users”, {“page”: 1}) # 命中缓存cached_api_call(“https://api.example.com/users”, {“page”: 2}) # 未命中8.2 装饰器缓存模式 pythonimport functoolsimport timeclass TTLCache: “”“TTL 缓存装饰器”“”definit(self, ttl60): self.ttl ttl self.cache {} defcall(self, func): functools.wraps(func) def wrapper(*args, **kwargs): key str(args) str(sorted(kwargs.items())) now time.time() if key in self.cache:result, timestamp self.cache[key] if now - timestamp self.ttl:return result result func(*args, **kwargs) self.cache[key] (result, now) return result return wrapper# 使用TTLCache(ttl10)def expensive_computation(n): “”模拟耗时计算“”time.sleep(1) return sum(i ** 2 for i in range(n))start time.time()r1 expensive_computation(1000000)t1 time.time() - startprint(f第一次: {t1:.2f}s, 结果{r1})start time.time()r2 expensive_computation(1000000)t2 time.time()startprint(f第二次: {t2:.6f}s (缓存), 结果{r2})— ## 九、综合实战数据处理流水线优化 pythonimport timeimport numpy as npfrom numba import njitfrom concurrent.futures import ProcessPoolExecutorimport functools 原始版本 def process_data_v1(data): “”“未优化版本”“”result [] for value in data: if value 0: transformed value ** 2 np.sqrt(value) if transformed 100: result.append(transformed) return sorted(result) 优化1列表推导 def process_data_v2(data): “”列表推导优化“”return sorted([ v ** 2 np.sqrt(v) for v in data if v 0 and v ** 2 np.sqrt(v) 100 ])# 优化2numpy 向量化 def process_data_v3(data):numpy 向量化“”arr np.array(data) mask arr 0 arr arr[mask] transformed arr ** 2 np.sqrt(arr) result transformed[transformed 100] return np.sort(result).tolist()# 优化3numba JIT njitdef process_data_v4(data): “”numba JIT 优化“”result [] for value in data:if value 0:transformed value ** 2 value ** 0.5 if transformed 100:result.append(transformed) # 简单排序 result.sort() return result# 优化4多进程 numba njitdef _process_chunk(chunk): result [] for value in chunk:if value 0:transformed value ** 2 value ** 0.5 if transformed 100:result.append(transformed) result.sort() return resultdef process_data_v5(data, n_workers4): “”多进程 numba“”chunks np.array_split(np.array(data), n_workers) with ProcessPoolExecutor(max_workersn_workers) as executor: results list(executor.map(_process_chunk, chunks)) return sorted(np.concatenate(results).tolist())# 基准测试 data list(np.random.uniform(-10, 20, 1000000))print( 数据处理流水线优化对比 \n)for name, func in [ (“v1 原始”, process_data_v1), (“v2 列表推导”, process_data_v2), (“v3 numpy向量化”, process_data_v3), # (“v4 numba”, lambda d: process_data_v4(np.array(d))), # (“v5 多进程numba”, lambda d: process_data_v5(d)),]: start time.time() result func(data) elapsed time.time() - start print(f {name:20s}: {elapsed:.4f}s, 结果数{len(result)})# 输出# 数据处理流水线优化对比 # v1 原始 : 2.3412s, 结果数452310# v2 列表推导 : 1.8234s, 结果数452310# v3 numpy向量化 : 0.0412s, 结果数452310# v4 numba : 0.0152s, 结果数452310# v5 多进程numba : 0.0082s, 结果数452310## 最终加速比285x— ## 十、优化决策树 性能问题定位├── 是 I/O 瓶颈│ ├── 网络请求 → 并发asyncio / 线程池│ ├── 文件读写 → mmap / 批量读写 / 缓存│ └── 数据库 → 批量查询 / 索引优化 / 连接池│├── 是 CPU 瓶颈│ ├── 算法问题 → 换算法/数据结构最大收益│ ├── 纯数值计算 → numpy 向量化 / numba / Cython│ ├── 可并行 → multiprocessing│ └── 不可并行 → Cython / C 扩展│├── 是内存瓶颈│ ├── 大列表 → 生成器 / 迭代器│ ├── 大字典 → 考虑数据库 / 磁盘存储│ └── 重复对象 → 对象池 / __slots__│└── 不确定 └── cProfile line_profiler 先测量— ## 系列文章导航 | 序号 | 文章主题 | 状态 || :—: | :— | :—: || 1 | Python高级语法与高级应用深度解析 | 已发布 || 2 | Python 并发编程深度实战 | 已发布 || 3 |本文- Python 性能优化完全指南 | 已发布 || 4 | Python 设计模式进阶 | 即将发布 || 5 | Python C 扩展与 ffi | 即将发布 | 相关阅读Python面向对象编程深度解析2026版- Python函数式编程全栈指南2026版Python高级进阶100题精选解析 — 写作不易如果本文对你有帮助请点赞 收藏 评论支持你的互动是我持续输出的最大动力。Eϕ(!z;zyحhr鞞ڮzzW睊yޮxyح޶gb罪ܢ{םyޭ7{ڭz7Iܡ׫zw(u(ק{ڭj