Cao Note

外观

Python性能优化 ​

性能优化的基本原则 ​

  1. 先测量，后优化：在优化之前，使用性能分析工具识别瓶颈
  2. 优化关键路径：优先优化执行频率高的代码
  3. 权衡优化成本：考虑优化的时间成本和收益
  4. 保持代码可读性：不要为了性能牺牲代码的可读性和可维护性
  5. 使用适当的算法和数据结构：选择时间复杂度和空间复杂度合适的算法
  6. 考虑硬件限制：了解CPU、内存、磁盘等硬件限制

性能分析工具 ​

内置性能分析工具 ​

time模块 ​

python
import time

start_time = time.time()
# 执行需要测量的代码
for i in range(1000000):
    pass
end_time = time.time()
print(f"执行时间: {end_time - start_time}秒")
import time

start_time = time.time()
# 执行需要测量的代码
for i in range(1000000):
    pass
end_time = time.time()
print(f"执行时间: {end_time - start_time}秒")

timeit模块 ​

python
import timeit

# 测量单行代码的执行时间
print(timeit.timeit('sum(range(100))', number=100000))

# 测量多行代码的执行时间
code = '''
sum = 0
for i in range(100):
    sum += i
'''
print(timeit.timeit(code, number=100000))
import timeit

# 测量单行代码的执行时间
print(timeit.timeit('sum(range(100))', number=100000))

# 测量多行代码的执行时间
code = '''
sum = 0
for i in range(100):
    sum += i
'''
print(timeit.timeit(code, number=100000))

cProfile模块 ​

python
import cProfile

# 测量函数的执行时间和调用次数
def test_function():
    sum = 0
    for i in range(1000000):
        sum += i
    return sum

cProfile.run('test_function()')
import cProfile

# 测量函数的执行时间和调用次数
def test_function():
    sum = 0
    for i in range(1000000):
        sum += i
    return sum

cProfile.run('test_function()')

第三方性能分析工具 ​

line_profiler ​

bash
pip install line_profiler
pip install line_profiler
python
# 在需要分析的函数上添加@profile装饰器
@profile
def test_function():
    sum = 0
    for i in range(1000000):
        sum += i
    return sum

test_function()
# 在需要分析的函数上添加@profile装饰器
@profile
def test_function():
    sum = 0
    for i in range(1000000):
        sum += i
    return sum

test_function()

运行命令：

bash
kernprof -l -v script.py
kernprof -l -v script.py

memory_profiler ​

bash
pip install memory_profiler
pip install memory_profiler
python
# 在需要分析的函数上添加@profile装饰器
@profile
def test_memory():
    data = [i for i in range(1000000)]
    return sum(data)

test_memory()
# 在需要分析的函数上添加@profile装饰器
@profile
def test_memory():
    data = [i for i in range(1000000)]
    return sum(data)

test_memory()

运行命令：

bash
python -m memory_profiler script.py
python -m memory_profiler script.py

代码优化技巧 ​

1. 使用局部变量 ​

局部变量的访问速度比全局变量快，因此在频繁访问的变量上，优先使用局部变量。

python
# 优化前
global_var = 10

def test_global():
    sum = 0
    for i in range(1000000):
        sum += global_var
    return sum

# 优化后
def test_local():
    local_var = 10
    sum = 0
    for i in range(1000000):
        sum += local_var
    return sum
# 优化前
global_var = 10

def test_global():
    sum = 0
    for i in range(1000000):
        sum += global_var
    return sum

# 优化后
def test_local():
    local_var = 10
    sum = 0
    for i in range(1000000):
        sum += local_var
    return sum

2. 避免在循环中使用点操作 ​

在循环中，避免使用对象的属性或方法调用，因为这会增加额外的开销。

python
# 优化前
import math
def test_dot():
    sum = 0
    for i in range(1000000):
        sum += math.sin(i)
    return sum

# 优化后
def test_optimized():
    sum = 0
    sin = math.sin
    for i in range(1000000):
        sum += sin(i)
    return sum
# 优化前
import math
def test_dot():
    sum = 0
    for i in range(1000000):
        sum += math.sin(i)
    return sum

# 优化后
def test_optimized():
    sum = 0
    sin = math.sin
    for i in range(1000000):
        sum += sin(i)
    return sum

3. 避免使用列表推导式生成大量数据 ​

当需要生成大量数据时，考虑使用生成器表达式，因为生成器表达式不会一次性生成所有数据，而是按需生成。

python
# 优化前
large_list = [i for i in range(10000000)]  # 占用大量内存

# 优化后
large_generator = (i for i in range(10000000))  # 占用很少内存
# 优化前
large_list = [i for i in range(10000000)]  # 占用大量内存

# 优化后
large_generator = (i for i in range(10000000))  # 占用很少内存

4. 使用向量化操作 ​

对于数值计算，优先使用NumPy的向量化操作，而不是Python的循环。

python
import numpy as np

# 优化前
def sum_python():
    sum = 0
    for i in range(1000000):
        sum += i
    return sum

# 优化后
def sum_numpy():
    arr = np.arange(1000000)
    return np.sum(arr)
import numpy as np

# 优化前
def sum_python():
    sum = 0
    for i in range(1000000):
        sum += i
    return sum

# 优化后
def sum_numpy():
    arr = np.arange(1000000)
    return np.sum(arr)

5. 使用内置函数和标准库 ​

Python的内置函数和标准库通常是用C实现的，执行速度比Python代码快得多。

python
# 优化前
def find_max_python(lst):
    max_val = lst[0]
    for val in lst:
        if val > max_val:
            max_val = val
    return max_val

# 优化后
def find_max_builtin(lst):
    return max(lst)
# 优化前
def find_max_python(lst):
    max_val = lst[0]
    for val in lst:
        if val > max_val:
            max_val = val
    return max_val

# 优化后
def find_max_builtin(lst):
    return max(lst)

6. 避免使用eval和exec ​

eval()和exec()函数会执行字符串形式的代码，执行速度慢且不安全，应尽量避免使用。

7. 使用f-strings进行字符串格式化 ​

f-strings是Python 3.6+引入的字符串格式化方式，比其他格式化方式更快。

python
name = "张三"
age = 20

# 优化前
print("姓名: %s, 年龄: %d" % (name, age))
print("姓名: {}, 年龄: {}".format(name, age))

# 优化后
print(f"姓名: {name}, 年龄: {age}")
name = "张三"
age = 20

# 优化前
print("姓名: %s, 年龄: %d" % (name, age))
print("姓名: {}, 年龄: {}".format(name, age))

# 优化后
print(f"姓名: {name}, 年龄: {age}")

数据结构优化 ​

1. 选择合适的数据结构 ​

数据结构插入/删除查找适用场景
列表O(1) - 末尾，O(n) - 其他位置O(n)有序数据，频繁访问
字典O(1)O(1)键值对存储，频繁查找
集合O(1)O(1)去重，成员关系检查
元组不可变O(n)不可变数据，频繁访问
双端队列O(1) - 两端O(n)频繁在两端插入/删除

2. 避免频繁的列表操作 ​

python
# 优化前
def concat_lists():
    result = []
    for i in range(1000):
        result = result + [i]  # 每次连接都会创建新列表
    return result

# 优化后
def append_list():
    result = []
    for i in range(1000):
        result.append(i)  # 原地修改，效率更高
    return result
# 优化前
def concat_lists():
    result = []
    for i in range(1000):
        result = result + [i]  # 每次连接都会创建新列表
    return result

# 优化后
def append_list():
    result = []
    for i in range(1000):
        result.append(i)  # 原地修改，效率更高
    return result

3. 使用生成器代替列表 ​

python
# 优化前
def generate_list():
    return [i for i in range(1000000)]  # 一次性生成所有数据

# 优化后
def generate_generator():
    return (i for i in range(1000000))  # 按需生成数据
# 优化前
def generate_list():
    return [i for i in range(1000000)]  # 一次性生成所有数据

# 优化后
def generate_generator():
    return (i for i in range(1000000))  # 按需生成数据

4. 使用collections模块 ​

python
from collections import defaultdict, deque, Counter

# defaultdict - 避免KeyError
d = defaultdict(int)
d["key"] += 1  # 无需检查键是否存在

# deque - 高效的双端队列
dq = deque()
dq.append(1)  # 右端插入
dq.appendleft(0)  # 左端插入
dq.pop()  # 右端删除
dq.popleft()  # 左端删除

# Counter - 计数
data = [1, 2, 3, 1, 2, 1, 3, 4, 5]
counter = Counter(data)  # {1: 3, 2: 2, 3: 2, 4: 1, 5: 1}
from collections import defaultdict, deque, Counter

# defaultdict - 避免KeyError
d = defaultdict(int)
d["key"] += 1  # 无需检查键是否存在

# deque - 高效的双端队列
dq = deque()
dq.append(1)  # 右端插入
dq.appendleft(0)  # 左端插入
dq.pop()  # 右端删除
dq.popleft()  # 左端删除

# Counter - 计数
data = [1, 2, 3, 1, 2, 1, 3, 4, 5]
counter = Counter(data)  # {1: 3, 2: 2, 3: 2, 4: 1, 5: 1}

内存优化 ​

1. 使用适当的数据类型 ​

python
import numpy as np

# 优化前
lst = [i for i in range(1000000)]  # 占用大量内存

# 优化后
arr = np.array([i for i in range(1000000)], dtype=np.int32)  # 使用较小的数据类型
import numpy as np

# 优化前
lst = [i for i in range(1000000)]  # 占用大量内存

# 优化后
arr = np.array([i for i in range(1000000)], dtype=np.int32)  # 使用较小的数据类型

2. 及时释放不再使用的变量 ​

python
# 优化前
def process_large_data():
    large_data = [i for i in range(1000000)]
    # 处理数据
    result = sum(large_data)
    return result

# 优化后
def process_large_data_optimized():
    large_data = [i for i in range(1000000)]
    # 处理数据
    result = sum(large_data)
    # 释放不再使用的变量
    del large_data
    return result
# 优化前
def process_large_data():
    large_data = [i for i in range(1000000)]
    # 处理数据
    result = sum(large_data)
    return result

# 优化后
def process_large_data_optimized():
    large_data = [i for i in range(1000000)]
    # 处理数据
    result = sum(large_data)
    # 释放不再使用的变量
    del large_data
    return result

3. 使用__slots__减少类实例的内存占用 ​

python
# 优化前
class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age

# 优化后
class Person:
    __slots__ = ['name', 'age']  # 限制实例的属性
    
    def __init__(self, name, age):
        self.name = name
        self.age = age
# 优化前
class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age

# 优化后
class Person:
    __slots__ = ['name', 'age']  # 限制实例的属性
    
    def __init__(self, name, age):
        self.name = name
        self.age = age

4. 使用生成器和迭代器 ​

python
# 优化前
def read_file(filename):
    with open(filename, 'r') as f:
        content = f.read()  # 一次性读取整个文件
        return content.split('\n')

# 优化后
def read_file_optimized(filename):
    with open(filename, 'r') as f:
        for line in f:  # 逐行读取文件
            yield line.strip()
# 优化前
def read_file(filename):
    with open(filename, 'r') as f:
        content = f.read()  # 一次性读取整个文件
        return content.split('\n')

# 优化后
def read_file_optimized(filename):
    with open(filename, 'r') as f:
        for line in f:  # 逐行读取文件
            yield line.strip()

并发和并行编程 ​

1. 多线程 ​

Python的多线程适合IO密集型任务，因为GIL（全局解释器锁）的存在，多线程不适合CPU密集型任务。

python
import threading
import time

def worker(num):
    print(f"线程 {num} 开始")
    time.sleep(2)  # 模拟IO操作
    print(f"线程 {num} 结束")

# 创建线程
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()
import threading
import time

def worker(num):
    print(f"线程 {num} 开始")
    time.sleep(2)  # 模拟IO操作
    print(f"线程 {num} 结束")

# 创建线程
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

2. 多进程 ​

Python的多进程适合CPU密集型任务，因为每个进程都有自己的Python解释器和内存空间，不受GIL的限制。

python
import multiprocessing
import time

def worker(num):
    print(f"进程 {num} 开始")
    # 模拟CPU密集型任务
    sum = 0
    for i in range(100000000):
        sum += i
    print(f"进程 {num} 结束")

# 创建进程
processes = []
for i in range(4):
    p = multiprocessing.Process(target=worker, args=(i,))
    processes.append(p)
    p.start()

# 等待所有进程完成
for p in processes:
    p.join()
import multiprocessing
import time

def worker(num):
    print(f"进程 {num} 开始")
    # 模拟CPU密集型任务
    sum = 0
    for i in range(100000000):
        sum += i
    print(f"进程 {num} 结束")

# 创建进程
processes = []
for i in range(4):
    p = multiprocessing.Process(target=worker, args=(i,))
    processes.append(p)
    p.start()

# 等待所有进程完成
for p in processes:
    p.join()

3. 异步编程 ​

Python的异步编程适合IO密集型任务，使用asyncio库。

python
import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://www.baidu.com",
        "https://www.google.com",
        "https://www.github.com"
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        print(f"获取了 {len(results)} 个网页")

asyncio.run(main())
import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://www.baidu.com",
        "https://www.google.com",
        "https://www.github.com"
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        print(f"获取了 {len(results)} 个网页")

asyncio.run(main())

4. 使用concurrent.futures模块 ​

python
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

def worker(num):
    print(f"任务 {num} 开始")
    time.sleep(2)
    print(f"任务 {num} 结束")
    return num

# 使用线程池
with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(worker, range(10)))
    print(f"结果: {results}")

# 使用进程池
with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(worker, range(10)))
    print(f"结果: {results}")
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

def worker(num):
    print(f"任务 {num} 开始")
    time.sleep(2)
    print(f"任务 {num} 结束")
    return num

# 使用线程池
with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(worker, range(10)))
    print(f"结果: {results}")

# 使用进程池
with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(worker, range(10)))
    print(f"结果: {results}")

第三方库和扩展 ​

1. Numba ​

Numba是一个JIT编译器，可以将Python代码编译为机器码，提高执行速度。

bash
pip install numba
pip install numba
python
from numba import jit
import time

# 使用@jit装饰器
@jit(nopython=True)
def sum_numbers(n):
    sum = 0
    for i in range(n):
        sum += i
    return sum

# 测量执行时间
start_time = time.time()
result = sum_numbers(100000000)
end_time = time.time()
print(f"执行时间: {end_time - start_time}秒")
from numba import jit
import time

# 使用@jit装饰器
@jit(nopython=True)
def sum_numbers(n):
    sum = 0
    for i in range(n):
        sum += i
    return sum

# 测量执行时间
start_time = time.time()
result = sum_numbers(100000000)
end_time = time.time()
print(f"执行时间: {end_time - start_time}秒")

2. Cython ​

Cython是一个Python的超集，可以将Python代码编译为C代码，提高执行速度。

bash
pip install cython
pip install cython

创建sum_numbers.pyx文件：

cython
def sum_numbers(int n):
    cdef int sum = 0
    cdef int i
    for i in range(n):
        sum += i
    return sum
def sum_numbers(int n):
    cdef int sum = 0
    cdef int i
    for i in range(n):
        sum += i
    return sum

创建setup.py文件：

python
from setuptools import setup
from Cython.Build import cythonize

setup(
    ext_modules=cythonize("sum_numbers.pyx")
)
from setuptools import setup
from Cython.Build import cythonize

setup(
    ext_modules=cythonize("sum_numbers.pyx")
)

编译和使用：

bash
python setup.py build_ext --inplace
python setup.py build_ext --inplace
python
import sum_numbers
result = sum_numbers.sum_numbers(100000000)
import sum_numbers
result = sum_numbers.sum_numbers(100000000)

3. PyPy ​

PyPy是一个Python解释器，使用JIT编译技术，执行速度比CPython快。

下载地址：https://www.pypy.org/download.html

使用方法：

bash
pypy script.py
pypy script.py

4. 使用C扩展 ​

对于性能要求极高的代码，可以使用C语言编写扩展模块。

数据库访问优化 ​

1. 使用连接池 ​

python
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

# 创建连接池
engine = create_engine(
    'mysql+pymysql://user:password@localhost/dbname',
    poolclass=QueuePool,
    pool_size=5,
    max_overflow=10
)

# 使用连接
with engine.connect() as conn:
    result = conn.execute("SELECT * FROM table")
    for row in result:
        print(row)
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

# 创建连接池
engine = create_engine(
    'mysql+pymysql://user:password@localhost/dbname',
    poolclass=QueuePool,
    pool_size=5,
    max_overflow=10
)

# 使用连接
with engine.connect() as conn:
    result = conn.execute("SELECT * FROM table")
    for row in result:
        print(row)

2. 批量操作 ​

python
import pymysql

# 连接数据库
conn = pymysql.connect(host='localhost', user='user', password='password', db='dbname')
cursor = conn.cursor()

# 批量插入
data = [(1, '张三'), (2, '李四'), (3, '王五')]
cursor.executemany("INSERT INTO users (id, name) VALUES (%s, %s)", data)
conn.commit()

# 关闭连接
cursor.close()
conn.close()
import pymysql

# 连接数据库
conn = pymysql.connect(host='localhost', user='user', password='password', db='dbname')
cursor = conn.cursor()

# 批量插入
data = [(1, '张三'), (2, '李四'), (3, '王五')]
cursor.executemany("INSERT INTO users (id, name) VALUES (%s, %s)", data)
conn.commit()

# 关闭连接
cursor.close()
conn.close()

3. 索引优化 ​

为查询频繁的列创建索引，提高查询速度。

4. 避免全表扫描 ​

sql
-- 优化前
SELECT * FROM users WHERE name = '张三';

-- 优化后（为name列创建索引）
CREATE INDEX idx_name ON users(name);
SELECT * FROM users WHERE name = '张三';
-- 优化前
SELECT * FROM users WHERE name = '张三';

-- 优化后（为name列创建索引）
CREATE INDEX idx_name ON users(name);
SELECT * FROM users WHERE name = '张三';

Web应用性能优化 ​

1. 使用缓存 ​

python
from flask import Flask
from flask_caching import Cache

app = Flask(__name__)
# 配置缓存
app.config['CACHE_TYPE'] = 'redis'
app.config['CACHE_REDIS_URL'] = 'redis://localhost:6379/0'
cache = Cache(app)

@app.route('/')
@cache.cached(timeout=60)  # 缓存60秒

def index():
    # 耗时操作
    return 'Hello, World!'
from flask import Flask
from flask_caching import Cache

app = Flask(__name__)
# 配置缓存
app.config['CACHE_TYPE'] = 'redis'
app.config['CACHE_REDIS_URL'] = 'redis://localhost:6379/0'
cache = Cache(app)

@app.route('/')
@cache.cached(timeout=60)  # 缓存60秒

def index():
    # 耗时操作
    return 'Hello, World!'

2. 使用CDN加速静态资源 ​

3. 压缩响应内容 ​

python
from flask import Flask
from flask_compress import Compress

app = Flask(__name__)
Compress(app)  # 启用响应压缩
from flask import Flask
from flask_compress import Compress

app = Flask(__name__)
Compress(app)  # 启用响应压缩

4. 使用异步Web框架 ​

如FastAPI、Tornado等，支持异步请求处理。

性能优化的最佳实践 ​

  1. 使用合适的算法和数据结构：选择时间复杂度和空间复杂度合适的算法
  2. 避免不必要的计算：缓存计算结果，避免重复计算
  3. 使用向量化操作：优先使用NumPy、Pandas等库的向量化操作
  4. 减少函数调用：避免在循环中频繁调用函数
  5. 使用局部变量：局部变量的访问速度比全局变量快
  6. 避免频繁的内存分配：使用对象池或预分配内存
  7. 使用多进程或异步编程：根据任务类型选择合适的并发模型
  8. 优化数据库访问：使用连接池、批量操作、索引优化等
  9. 使用性能分析工具：定期使用性能分析工具识别瓶颈
  10. 持续优化：性能优化是一个持续的过程，需要定期检查和优化

总结 ​

Python性能优化是一个复杂的话题，需要从多个方面入手。在优化之前，首先要使用性能分析工具识别瓶颈，然后根据瓶颈的类型选择合适的优化方法。

对于不同类型的任务，需要选择不同的优化策略：

  • IO密集型任务：使用多线程、异步编程或协程
  • CPU密集型任务：使用多进程、Cython、Numba或PyPy
  • 内存密集型任务：优化数据结构、减少内存占用、及时释放变量

同时，也要考虑代码的可读性和可维护性，不要为了性能牺牲代码的质量。

通过合理的性能优化，可以显著提高Python代码的执行速度和资源利用率，使Python应用能够处理更大规模的数据和更高并发的请求。

Copyright © 2024-present xiaocao