关闭

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

发表于:2024-3-29 09:35

字体: | 上一篇 | 下一篇 | 我要投稿

 作者:郭小喵学AI    来源:小喵学AI

  Python是目前世界上增长最快的编程语言之一,深受全球开发者的喜爱。其简单语法和丰富的库使得在各个领域都能得到广泛应用,比如数据科学、机器学习、信号处理、数据可视化等。然而,Python在解决复杂问题时可能会显得执行速度较慢。因此,本文将探讨一些优化Python代码的方法,以加速代码运行。
  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中执行的某些计算密集型任务得以显著加速。
  本文内容不用于商业目的,如涉及知识产权问题,请权利人联系51Testing小编(021-64471599-8017),我们将立即处理
《2023软件测试行业现状调查报告》独家发布~

关注51Testing

联系我们

快捷面板 站点地图 联系我们 广告服务 关于我们 站长统计 发展历程

法律顾问:上海兰迪律师事务所 项棋律师
版权所有 上海博为峰软件技术股份有限公司 Copyright©51testing.com 2003-2024
投诉及意见反馈:webmaster@51testing.com; 业务联系:service@51testing.com 021-64471599-8017

沪ICP备05003035号

沪公网安备 31010102002173号