Python案例如何优化CPU占用?

wen python案例 67

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

Python案例如何优化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空闲,但进程不退让。 优化:用asyncioconcurrent.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.Poolconcurrent.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%的改进效果。

抱歉,评论功能暂时关闭!