优化Python程序的CPU占用,核心思路是减少不必要的计算、让CPU在等待时能休息以及利用多核并行,以下按场景分类,从易到难给出具体案例和优化方法。

案例1:死循环或忙等待(最常见)
问题:while True: pass 或不断轮询检查某个条件会占满CPU。
优化:加入休眠或改为事件驱动。
# ❌ 忙等待(占满CPU)
import time
flag = False
while not flag:
# 持续空转检查
pass
# ✅ 加入休眠
while not flag:
time.sleep(0.1) # 每100ms检查一次,CPU占用从100%降到接近0%
# ✅ 更优雅:使用 threading.Event 或 asyncio 事件
import threading
event = threading.Event()
# 在另一线程或回调中 event.set()
event.wait() # 阻塞等待,不占CPU
优化效果:CPU占用从100%降至<1%。
案例2:大量字符串拼接(内存与CPU双重浪费)
问题:在循环中使用 拼接字符串,会产生大量临时对象,触发频繁GC和拷贝。
优化:改用 str.join() 或列表推导式。
# ❌ 不推荐
result = ""
for i in range(100000):
result += str(i) + ","
# ✅ 推荐
result = ",".join(str(i) for i in range(100000))
优化效果:时间从O(n²)降为O(n),CPU占用下降90%以上。
案例3:使用纯Python计算代替NumPy/SciPy
问题:对大规模数值计算逐元素循环(如矩阵乘法、图像处理)。 优化:用NumPy向量化操作,底层用C/Fortran执行。
import numpy as np
import time
# ❌ 纯Python循环(慢)
def py_sum(arr):
total = 0.0
for x in arr:
total += x
return total
data = [i * 0.1 for i in range(10_000_000)]
t0 = time.time()
py_sum(data) # 耗时 ~0.8秒
# ✅ NumPy向量化(快)
arr = np.array(data)
t0 = time.time()
np.sum(arr) # 耗时 ~0.02秒
优化效果:速度提升几十倍,CPU占用大幅降低。
案例4:爬虫/IO密集型任务使用多线程/异步
问题:单线程爬虫等待网络IO时CPU空闲,但进程不退让。
优化:用asyncio或concurrent.futures.ThreadPool并发等待。
import requests
import time
urls = ["http://example.com"] * 100
# ❌ 同步串行(CPU在等待IO时也占着)
t0 = time.time()
for url in urls:
requests.get(url)
print(f"同步耗时: {time.time()-t0:.2f}s")
# ✅ 异步(使用aiohttp,不阻塞CPU)
import aiohttp
import asyncio
async def fetch(session, url):
async with session.get(url) as resp:
return await resp.text()
async def main():
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, url) for url in urls]
await asyncio.gather(*tasks)
t0 = time.time()
asyncio.run(main())
print(f"异步耗时: {time.time()-t0:.2f}s")
优化效果:并发数量增加时,CPU占用不高但吞吐量提升10~100倍。
案例5:CPU密集型任务使用多进程(而非多线程)
问题:Python GIL导致多线程无法利用多核,CPU占用看似高但实际效率低。
优化:用multiprocessing.Pool或concurrent.futures.ProcessPool。
import math
import multiprocessing as mp
import time
# ❌ 多线程计算(受GIL限制,CPU占用高但实际慢)
# ✅ 多进程(每个进程独立Python解释器,真正并行)
def is_prime(n):
if n < 2:
return False
for i in range(2, int(math.sqrt(n)) + 1):
if n % i == 0:
return False
return True
numbers = list(range(10_000_000, 10_010_000))
# 单进程
t0 = time.time()
results = [is_prime(n) for n in numbers]
print(f"单进程: {time.time()-t0:.2f}s")
# 多进程(使用所有CPU核)
with mp.Pool(processes=mp.cpu_count()) as pool:
t0 = time.time()
results = pool.map(is_prime, numbers)
print(f"多进程: {time.time()-t0:.2f}s")
优化效果:在4核机器上,时间缩短为单进程的1/4左右,CPU利用率达到400%。
案例6:频繁函数调用/循环内嵌套函数
问题:循环内反复调用开销大的函数,或使用lambda/eval。
优化:将函数提到循环外,或使用局部变量绑定。
# ❌ 每次循环都要查找全局变量
global_var = 1
def slow():
for i in range(10_000):
_ = global_var + i
# ✅ 局部变量绑定
def fast():
local_var = global_var
for i in range(10_000):
_ = local_var + i
# 使用timeit测试可发现fast比slow快20~30%
优化效果:循环规模大时,可节省20~50% CPU时间。
案例7:使用__slots__减少内存访问开销
问题:大量自定义对象(如游戏实体、数据记录)使用__dict__存储属性,占用内存并降低访问速度。
优化:定义__slots__,Python会使用紧凑的数组存储属性,减少内存碎片和属性查找开销。
# ❌ 普通类
class Point:
def __init__(self, x, y):
self.x = x
self.y = y
# ✅ 使用__slots__
class Point:
__slots__ = ('x', 'y')
def __init__(self, x, y):
self.x = x
self.y = y
# 实例化100万个对象,__slots__版本内存减少约40%,访问速度提升约15%
案例8:使用line_profiler/cProfile定位瓶颈
问题:盲目优化而不测量,可能浪费精力。
优化:先用cProfile找出热点函数,再针对性优化。
python -m cProfile -s cumulative my_script.py
或使用py-spy实时监控CPU占用:
pip install py-spy py-spy record -o profile.svg --pid <进程ID> # 生成火焰图
针对不同场景的优化优先级
| 场景 | 首选优化 | 效果 |
|---|---|---|
| 死循环/轮询 | 加sleep或事件驱动 | 极大(CPU 100%→0%) |
| 数值计算 | NumPy / Numba / C扩展 | 10~100倍 |
| 网络IO密集 | asyncio / 多线程 | 10~100倍吞吐 |
| CPU密集(有GIL) | multiprocessing | 核数倍加速 |
| 字符串/列表操作 | join / 列表推导 / map | 10~100倍 |
| 大量小对象 | slots / 使用array模块 | 2~5倍 |
| 循环内查全局 | 局部变量绑定 | 3~2倍 |
| 定位瓶颈 | 先profiling再优化 | 避免浪费精力 |
最后建议:不要过早优化,先用 time.time() 或 cProfile 确认热点,然后针对最耗时的1~2个点优化,通常能获得80%的改进效果。