okgoes
🔍
← 目录从零开始学python
✏️ 编辑

并行与异步编程（asyncio, concurrent.futures）

2024/8/19

并行与异步编程在 Python 中是提高程序性能、减少等待时间的关键技术。它们各自有不同的应用场景，适用于处理 I/O 密集型任务或 CPU 密集型任务。

并行编程

并行编程允许多个任务同时执行，利用多核处理器的能力。Python 提供了 concurrent.futures 模块来方便地管理并行任务。

concurrent.futures 模块

concurrent.futures 模块提供了两个执行器（Executor）类：

  • ThreadPoolExecutor: 用于 I/O 密集型任务，通过线程池实现并发。
  • ProcessPoolExecutor: 用于 CPU 密集型任务，通过进程池实现并行。
示例：使用 ThreadPoolExecutor

适用于 I/O 密集型任务，如网络请求、文件操作等。

python
import concurrent.futures
import time

def fetch_data(index):
    print(f"Fetching data for {index}...")
    time.sleep(2)  # 模拟I/O操作的延迟
    return f"Data {index}"

# 使用 ThreadPoolExecutor 进行并发任务执行
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(fetch_data, range(10)))

for result in results:
    print(result)
示例：使用 ProcessPoolExecutor

适用于 CPU 密集型任务，如计算、图像处理等。

python
import concurrent.futures
import math

def compute_factorial(n):
    return math.factorial(n)

numbers = [5, 10, 20, 30, 50]

# 使用 ProcessPoolExecutor 进行并行计算
with concurrent.futures.ProcessPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(compute_factorial, numbers))

for number, result in zip(numbers, results):
    print(f"Factorial of {number} is {result}")

异步编程

异步编程允许程序在等待 I/O 操作时执行其他任务，从而提高程序的响应速度。Python 的 asyncio 模块是异步编程的核心模块。

asyncio 模块

asyncio 模块提供了 async 和 await 关键字，用于定义和执行异步函数。

示例：使用 asyncio 进行异步 I/O 操作
python
import asyncio

async def fetch_data(index):
    print(f"Fetching data for {index}...")
    await asyncio.sleep(2)  # 模拟I/O操作的延迟
    return f"Data {index}"

async def main():
    tasks = [fetch_data(i) for i in range(5)]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)

# 运行异步主函数
asyncio.run(main())

何时使用并行 vs 异步

  • 并行编程：适合 CPU 密集型任务，需要利用多核处理器的能力。
  • 异步编程：适合 I/O 密集型任务，如网络请求、文件读取等，不需要大量 CPU 运算，主要是减少等待时间。

组合使用 asyncio 和 concurrent.futures

你可以将 concurrent.futures 与 asyncio 结合使用，在异步函数中使用并行计算：

python
import asyncio
import concurrent.futures

def compute_factorial(n):
    return math.factorial(n)

async def main():
    loop = asyncio.get_running_loop()
  
    with concurrent.futures.ProcessPoolExecutor() as executor:
        numbers = [5, 10, 20, 30, 50]
        tasks = [
            loop.run_in_executor(executor, compute_factorial, number)
            for number in numbers
        ]
        results = await asyncio.gather(*tasks)
  
    for number, result in zip(numbers, results):
        print(f"Factorial of {number} is {result}")

# 运行异步主函数
asyncio.run(main())

在这个示例中，我们结合使用了 asyncio 和 ProcessPoolExecutor，在异步函数中执行 CPU 密集型任务，充分利用多核 CPU，同时保持异步的优势。