4

加速 Python 代码的八个优秀实用技巧

 1 month ago
source link: https://www.51cto.com/article/784734.html
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.

加速 Python 代码的八个优秀实用技巧

作者:郭小喵学AI 2024-03-27 14:06:58
Python在解决复杂问题时可能会显得执行速度较慢,因此,本文将探讨一些优化Python代码的方法,以加速代码运行。

Python是目前世界上增长最快的编程语言之一,深受全球开发者的喜爱。其简单语法和丰富的库使得在各个领域都能得到广泛应用,比如数据科学、机器学习、信号处理、数据可视化等。然而,Python在解决复杂问题时可能会显得执行速度较慢。因此,本文将探讨一些优化Python代码的方法,以加速代码运行。

1783ad339e7a0ed493f680e1e0bf980fba21bc.jpg

1.使用内置函数和库

Python标准库和第三方库(如NumPy、Pandas等)中的函数通常是用C或Cython编写的,运行速度远超纯Python代码。为了加速Python代码,可以尽量使用这些库中的向量化操作代替Python原生循环,特别是在处理数组和矩阵运算时。

举个例子,计算Python列表中每个元素的平方。

import numpy as np
import time

# 定义一个Python列表
python_list = [1, 2, 3, 4, 5]

# 使用纯Python循环计算平方的时间测试
def measure_time(function, argument):
    start_time = time.time()
    result = function(argument)
    end_time = time.time()
    return result, end_time - start_time

# 定义纯Python循环计算平方的函数
def square_elements_python(lst):
    squared_lst = []
    for num in lst:
        squared_lst.append(num ** 2)
    return squared_lst

# 计算并输出纯Python循环方法执行时间和结果
python_squares, python_time = measure_time(square_elements_python, python_list)
print(f"纯Python循环方法: {python_squares}, Time taken: {python_time} seconds")

# 转换为NumPy数组并使用向量化操作
start_time = time.time()
numpy_array = np.array(python_list)
numpy_squares = numpy_array ** 2
end_time = time.time()

# 输出NumPy向量化操作执行时间
numpy_time = end_time - start_time
print(f"NumPy向量化操作: {numpy_squares.tolist()}, Time taken: {numpy_time} seconds")

输出结果如下,由此可以看出NumPy的向量化操作在执行速度上远超纯Python循环法。这是因为NumPy内部对数组操作进行了高度优化,并利用C语言编写的底层算法,极大地提高了处理效率。

纯Python循环方法: [1, 4, 9, 16, 25], Time taken: 4.5299530029296875e-06 seconds
NumPy向量化操作: [1, 4, 9, 16, 25], Time taken: 0.00020122528076171875 seconds

2.Numba JIT编译

可以使用Numba库进行即时(JIT)编译,它可以将指定的Python函数转换为高效机器码,以提升执行速度。尤其适用于数值计算密集型代码。

例如下面的代码,sum_array函数被装饰器@jit(nopython=True)标记后,Numba会对其进行即时编译,将其转换为机器码以提升计算密集型任务的执行速度。

import numpy as np
from numba import jit

@jit(nopython=True)
def sum_array(arr: np.ndarray) -> float:
    result = 0.0
    for i in range(arr.shape[0]):
        result += arr[i]
    return result

arr = np.array([1.0, 2.0, 3.0, 4.0, 5.0])
print(sum_array(arr))

3.避免不必要的copy操作

尽可能在原地修改对象,而不是创建新对象。例如,使用列表的extend()方法而非"+"操作符进行合并,使用numpy数组的切片赋值而不是重新创建数组。例如:

# 避免:
list1 = [1, 2, 3]
list2 = [4, 5, 6]
new_list = list1 + list2

# 推荐:
list1 = [1, 2, 3]
list2 = [4, 5, 6]
# 这里不会创建新的列表对象,而是直接在原地扩展list1
list1.extend(list2)

4.使用生成器表达式代替列表推导

当不需要一次性生成所有结果,而是逐个处理时,使用生成器表达式代替列表推导式可以节省内存,因为它不会立即创建完整列表。

例如假设有一个包含整数的列表,我们想要计算每个整数的平方并输出结果。使用列表推导式的方法如下:

numbers = [1, 2, 3, 4, 5]
squared_numbers = [num ** 2 for num in numbers]
for squared_num in squared_numbers:
    print(squared_num)

输出结果为:

1
4
9
16
25

但是,如果我们只需要逐个处理每个平方数,而不需要将它们存储在内存中,可以使用生成器表达式代替列表推导式:

numbers = [1, 2, 3, 4, 5]
squared_numbers = (num ** 2 for num in numbers)
for squared_num in squared_numbers:
    print(squared_num)

输出结果与之前相同,但是使用生成器表达式可以节省内存,因为它不会一次性生成所有结果,而是逐个生成。

5.合理利用多线程或多进程

对于CPU密集型任务,Python的多线程受GIL限制,但对于IO密集型任务或是使用多核CPU处理CPU密集型任务时,可以通过multiprocessing库开启多进程来提升效率。

例如如下代码定义了一个计算密集型函数cpu_bound_task,然后通过multiprocessing.Pool创建了与CPU核心数量相等的进程池,并用pool.map()方法将输入列表中的任务分配给这些进程进行并行处理。

这样,每个进程都有自己的内存空间和独立GIL,从而可以充分利用多核处理器的能力提高执行效率。

import multiprocessing as mp

def cpu_bound_task(n):
    # 模拟的CPU密集型计算任务
    result = 0
    for i in range(n):
        result += i * i
    return result

if __name__ == "__main__":
    inputs = [1_000_000 + x for x in range(10)]  # 多个需要处理的数据单元

    with mp.Pool(processes=mp.cpu_count()) as pool:  # 使用所有可用CPU核心数
        results = pool.map(cpu_bound_task, inputs)  # 将任务分配到各个进程中并行处理

    print(f"Results: {results}")

6.缓存计算结果

如果存在重复计算的情况,可以使用functools.lru_cache装饰器来缓存函数的返回结果,避免重复计算。

如下示例使用Python标准库中的functools.lru_cache装饰器来缓存函数的结果,避免重复计算。

from functools import lru_cache

@lru_cache(maxsize=None)  # 缓存所有结果,可以根据实际情况设置缓存大小
def expensive_computation(x):
    # 假设这是一个计算成本很高的函数
    print("Computing...")
    return x ** x

# 第一次调用时会执行计算
result1 = expensive_computation(5)
# 第二次调用时会从缓存中获取结果,不再执行计算
result2 = expensive_computation(5)
print(result1 == result2)

第一次调用expensive_computation(5)时,执行计算并打印"Computing...",然后返回计算结果25。第二次调用时,由于结果已被缓存,不再执行计算,直接返回上次计算得到的25。因此,result1 == result2的输出是True。

7.利用异步IO

在处理大量IO操作时,如网络请求、文件读写等,可以利用asyncio库实现异步编程,最大化利用等待IO完成的时间进行其他任务的处理。

例如下面例子使用Python的asyncio库来并行处理多个网络请求,它会同时发起10个对'http://example.com'的异步网络请求,并等待所有请求完成后,通过responses变量获取所有的响应结果,然后逐个调用process_response(response)函数处理这些响应。

import asyncio

async def fetch(url):
    # 异步网络请求
    response = await asyncio.get_event_loop().run_in_executor(None, fetch_from_network, url)
    return response

async def main():
    tasks = [fetch('http://example.com') for _ in range(10)]
    responses = await asyncio.gather(*tasks)
    for response in responses:
        process_response(response)

# 启动事件循环
asyncio.run(main())

8.使用Cython或者Python-C接口

对于计算密集型的部分代码,可以使用Cython编写,将其编译为C扩展模块,或者直接使用Python的C API编写扩展模块,这种方式可以大幅提高这部分代码的执行效率。示例如下:

首先,安装Cython并创建.pyx文件:

# example_cython.pyx
def cython_power(int x):
    return x ** x

然后,编译为C扩展模块:

$ cython example_cython.pyx
$ gcc -shared -pthread -fPIC -fwrapv -O2 -Wall -fno-strict-aliasing -I/usr/include/python3.7 -o example_cython.cpython-37m-x86_64-linux-gnu.so example_cython.c

最后,在Python中导入并使用:

import example_cython

# 使用Cython优化后的函数
result = example_cython.cython_power(5)
print(result)

通过这种方法,Cython能够自动将Python代码转化为C代码,使得原本在Python中执行的某些计算密集型任务得以显著加速。

责任编辑:赵宁宁 来源: 小喵学AI

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK