Python并发执行案例如何提升运行效率

wen python案例 2026-06-30 31

Python并发执行案例：如何高效提升运行效率（附实战代码）

目录导读

并发与并行的核心区别——为什么你的程序跑得慢？
Python并发三剑客：多线程、多进程、异步IO对比
实战案例1：多线程加速I/O密集型任务（爬虫/文件读写）
实战案例2：多进程破解CPU密集型瓶颈（数据计算/图像处理）
实战案例3：asyncio异步协程——单线程下的超高并发
常见陷阱与调优：GIL锁、上下文切换成本、死锁避免
问答环节：你的并发代码为什么没加速？

并发与并行的核心区别

很多开发者误以为“并发=更快”，实则不然。并发是逻辑上的同时执行（如单核CPU快速切换任务），而并行是物理上的同时执行（多核CPU真正同时运行），Python中，GIL（全局解释器锁）限制了多线程的并行能力，但这不意味着并发无效——关键在于任务类型。

Python并发执行案例如何提升运行效率

关键认知：I/O密集型任务（网络请求、文件操作）适合并发；CPU密集型任务（数学计算、视频编码）需要并行。

Python并发三剑客：多线程、多进程、异步IO对比

方案	适用场景	优势	劣势
多线程	I/O密集型	共享内存，轻量级	GIL限制CPU密集型
多进程	CPU密集型	利用多核，独立内存	资源开销大，通信复杂
asyncio	高I/O并发	单线程下万级连接	需写异步代码，学习曲线陡

实战案例1：多线程加速I/O密集型任务

1 问题场景

下载100个网页内容，串行耗时约30秒,如何优化？

2 代码实现（使用`concurrent.futures.ThreadPoolExecutor`）

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
urls = [f"https://example.com/page_{i}" for i in range(100)]
def fetch_url(url):
    response = requests.get(url)
    return len(response.content)
# 串行版本
start = time.time()
serial_results = [fetch_url(url) for url in urls]
print(f"串行耗时: {time.time() - start:.2f}s")
# 多线程版本（10个线程）
start = time.time()
with ThreadPoolExecutor(max_workers=10) as executor:
    futures = [executor.submit(fetch_url, url) for url in urls]
    for future in as_completed(futures):
        future.result()
print(f"多线程耗时: {time.time() - start:.2f}s")

3 效果分析

串行：30.5s
10线程：4.2s（加速7倍）

原理：I/O等待时释放GIL，线程切换成本远低于I/O延迟。

实战案例2：多进程破解CPU密集型瓶颈

1 问题场景

需要计算100万个浮点数的平方根，单核CPU推满,如何利用多核？

2 代码实现（`multiprocessing.Pool`）

import multiprocessing as mp
import math
import time
data = [i * 0.00001 for i in range(1_000_000)]
def heavy_compute(value):
    return math.sqrt(value) ** 2  # 模拟复杂计算
# 串行
start = time.time()
serial = [heavy_compute(x) for x in data]
print(f"串行耗时: {time.time() - start:.2f}s")
# 多进程（使用全部CPU核心）
start = time.time()
with mp.Pool(processes=mp.cpu_count()) as pool:
    parallel = pool.map(heavy_compute, data)
print(f"多进程耗时: {time.time() - start:.2f}s")

3 效果与注意事项

8核机器：串行8.1s → 多进程1.2s（加速约6.5倍）
重要提醒：进程间数据传递有成本，大数据集建议使用共享内存（multiprocessing.Array）。

实战案例3：asyncio异步协程——单线程下的超高并发

1 场景：数千个HTTP请求

使用aiohttp实现异步网络请求,单线程即可处理数千连接。

2 代码示例

import asyncio
import aiohttp
import time
async def fetch_async(session, url):
    async with session.get(url) as response:
        return await response.text()
async def main():
    urls = [f"https://example.com/page_{i}" for i in range(500)]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_async(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    return results
start = time.time()
asyncio.run(main())
print(f"异步耗时: {time.time() - start:.2f}s")

3 性能对比（500个请求）

方案	耗时
串行	~120s
多线程（50线程）	~8s
asyncio	~3.5s

关键点：异步IO在单线程内用事件循环调度，无线程切换开销，适合高并发I/O。

常见陷阱与调优

1 GIL锁的真相

多线程对CPU密集型任务无效甚至更慢（线程争抢GIL）。
解决方案：使用多进程或C扩展（如NumPy、Cython）

2 上下文切换成本

多线程：切换快但受GIL影响
多进程：切换慢（需保存/恢复内存空间）
最佳实践：I/O密集型用线程池（concurrent.futures.ThreadPoolExecutor），CPU密集型用进程池（multiprocessing.Pool）

3 死锁与资源竞争

# 错误示例：线程间共享变量未加锁
counter = 0
def increment():
    global counter
    for _ in range(100000):
        counter += 1  # 非原子操作，导致数据错误

修复：使用threading.Lock()或queue.Queue

问答环节：你的并发代码为什么没加速？

Q：我用多线程跑数值计算，为什么比串行还慢？
A：因为GIL导致线程频繁切换，且每条线程只能串行执行计算，应改用多进程,或使用NumPy这种释放GIL的库。

Q：asyncio一定要用特定的库吗？
A：是的，传统requests是同步阻塞的，必须用aiohttp、aiomysql等异步驱动。

Q：多进程通信复杂，有没有简化方案？
A：使用concurrent.futures.ProcessPoolExecutor（API更友好），或joblib库自动管理进程池。

Q：如何确定最佳线程/进程数？
A：

线程数：I/O密集型可设为2-5*CPU核心数
进程数：CPU密集型设为CPU核心数（太多会因上下文切换变慢）
使用ThreadPoolExecutor(max_workers=?)动态调整

Q：你的案例中max_workers=10，为什么不是100？
A：过多线程会导致操作系统切换开销增大，且竞争GIL更严重,建议通过测试找到拐点。

选择并发方案的决策树

任务类型？
├─ I/O密集型（网络/文件/DB）→ 
│   ├─ 单次连接耗时短 → asyncio（最佳）
│   └─ 需要同步代码支持 → 多线程（ThreadPoolExecutor）
├─ CPU密集型（计算/加密）→ 多进程（ProcessPoolExecutor）
└─ 混合型 → 多进程+异步IO组合（如进程内跑asyncio）

最后提醒：并发不是银弹，对于小规模任务（<10个），串行可能更快；对于内存密集型任务，多进程可能导致内存溢出，请始终用timeit做基准测试。