本文目录导读：
我来分享几个Python并发优化的实战案例,从简单到复杂逐步深入。
基础案例：文件批量处理优化
原始版本（同步处理）
import requests
from time import sleep
def download_file(url):
print(f"下载: {url}")
sleep(1) # 模拟IO操作
return f"完成: {url}"
# 同步处理100个文件需要100秒
urls = [f"http://example.com/file_{i}" for i in range(100)]
for url in urls:
download_file(url)
优化版本（线程池）
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
def download_file(url):
time.sleep(1)
return f"完成: {url}"
urls = [f"http://example.com/file_{i}" for i in range(100)]
# 使用线程池，设置合理的工作线程数
with ThreadPoolExecutor(max_workers=10) as executor:
# 提交所有任务
future_to_url = {executor.submit(download_file, url): url for url in urls}
# 处理完成的任务
for future in as_completed(future_to_url):
url = future_to_url[future]
try:
result = future.result()
print(result)
except Exception as e:
print(f"{url} 失败: {e}")
优化效果：从100秒降到约10秒（10个线程并行）
进阶案例：爬虫数据采集优化
原始版本（串行+简单并发）
import requests
def fetch_page(url):
resp = requests.get(url)
return resp.text
# 简单粗暴的并发，可能导致内存暴涨
urls = [...] # 大量URL
with ThreadPoolExecutor(max_workers=100) as executor:
results = list(executor.map(fetch_page, urls)) # 一次性加载所有结果
优化版本（生产者-消费者模式）
import asyncio
import aiohttp
from asyncio import Semaphore
from collections import deque
import aiofiles
class AsyncCrawler:
def __init__(self, max_concurrent=20):
self.semaphore = Semaphore(max_concurrent)
self.session = None
async def fetch_with_limit(self, url):
async with self.semaphore: # 限制并发数
async with self.session.get(url) as response:
return await response.text()
async def process_url(self, url):
try:
html = await self.fetch_with_limit(url)
# 异步写入文件，避免IO阻塞
filename = f"data/{url.split('/')[-1]}.html"
async with aiofiles.open(filename, 'w') as f:
await f.write(html)
return True
except Exception as e:
print(f"失败: {url}, {e}")
return False
async def batch_process(self, urls, batch_size=50):
connector = aiohttp.TCPConnector(limit=30) # 限制连接池大小
async with aiohttp.ClientSession(connector=connector) as session:
self.session = session
# 分批处理，避免内存溢出
for i in range(0, len(urls), batch_size):
batch = urls[i:i+batch_size]
tasks = [self.process_url(url) for url in batch]
results = await asyncio.gather(*tasks)
# 实时统计
success = sum(results)
print(f"批次 {i//batch_size + 1}: {success}/{len(batch)} 成功")
# 使用
async def main():
urls = [f"http://example.com/page_{i}" for i in range(1000)]
crawler = AsyncCrawler(max_concurrent=30)
await crawler.batch_process(urls)
asyncio.run(main())
高性能案例：数据处理管道
使用多进程+协程混合架构
import asyncio
from multiprocessing import Process, Queue, Event
import numpy as np
from queue import Empty
class DataPipeline:
"""数据流水线：CPU密集型任务用多进程，IO密集型用协程"""
def __init__(self, num_workers=4, buffer_size=1000):
self.input_queue = Queue(maxsize=buffer_size)
self.output_queue = Queue(maxsize=buffer_size)
self.num_workers = num_workers
self.stop_event = Event()
async def fetcher(self, source):
"""异步数据获取器"""
async for data in source:
# 异步获取数据
processed = await self.preprocess(data)
# 放入队列，阻塞如果队列满
self.input_queue.put(processed, block=True)
def processor(self, worker_id):
"""CPU密集型数据处理（多进程）"""
while not self.stop_event.is_set():
try:
data = self.input_queue.get(timeout=1)
# CPU密集型计算
result = self.heavy_computation(data)
self.output_queue.put(result)
except Empty:
continue
def heavy_computation(self, data):
"""模拟复杂计算"""
# 使用numpy进行向量化计算
array = np.array(data)
result = np.fft.fft(array) # FFT变换
return np.abs(result).mean()
async def writer(self):
"""异步结果写入"""
results = []
while not self.stop_event.is_set() or not self.output_queue.empty():
try:
result = self.output_queue.get_nowait()
results.append(result)
if len(results) >= 100:
# 批量异步写入
await self.batch_write(results)
results.clear()
except Empty:
await asyncio.sleep(0.1)
# 写入剩余数据
if results:
await self.batch_write(results)
async def batch_write(self, data):
"""批量写入（模拟）"""
await asyncio.sleep(0.01)
# 实际写入数据库或文件
def run(self, source):
"""启动流水线"""
# 启动多个处理进程
workers = []
for i in range(self.num_workers):
p = Process(target=self.processor, args=(i,))
p.start()
workers.append(p)
# 运行异步任务
loop = asyncio.get_event_loop()
tasks = [
self.fetcher(source),
self.writer()
]
loop.run_until_complete(asyncio.gather(*tasks))
# 清理
self.stop_event.set()
for w in workers:
w.join()
# 使用
async def data_source():
"""异步数据源"""
for i in range(1000000):
yield np.random.random(1000)
await asyncio.sleep(0.001) # 模拟网络延迟
pipeline = DataPipeline(num_workers=4)
pipeline.run(data_source())
实际案例：Web API服务优化
使用异步Web框架
from fastapi import FastAPI, BackgroundTasks
import asyncio
from typing import Dict
import time
app = FastAPI()
# 缓存层
class Cache:
def __init__(self):
self._cache: Dict[str, bytes] = {}
self._lock = asyncio.Lock()
async def get(self, key: str):
async with self._lock:
return self._cache.get(key)
async def set(self, key: str, value: bytes, ttl: int = 300):
async with self._lock:
self._cache[key] = value
# 异步删除过期缓存
asyncio.create_task(self._expire_after(key, ttl))
async def _expire_after(self, key: str, ttl: int):
await asyncio.sleep(ttl)
async with self._lock:
self._cache.pop(key, None)
cache = Cache()
# 请求合并优化
class RequestBatcher:
"""合并相同请求，减少数据库查询"""
def __init__(self, max_batch_size=50, wait_time=0.01):
self.max_batch_size = max_batch_size
self.wait_time = wait_time
self._batch = []
self._lock = asyncio.Lock()
async def add_request(self, user_id: int) -> dict:
"""添加请求并等待批量处理"""
future = asyncio.get_event_loop().create_future()
async with self._lock:
self._batch.append((user_id, future))
current_batch = self._batch # 引用当前批次
if len(self._batch) >= self.max_batch_size:
self._batch = [] # 创建新批次
asyncio.create_task(self._process_batch(current_batch))
# 如果是批次中的第一个，设置定时器
if len(current_batch) == 1:
asyncio.create_task(self._delayed_process(current_batch))
return await future
async def _delayed_process(self, batch):
"""延迟处理，等待更多请求加入"""
await asyncio.sleep(self.wait_time)
async with self._lock:
if self._batch is batch: # 仍然是当前批次
self._batch = []
asyncio.create_task(self._process_batch(batch))
async def _process_batch(self, batch):
"""批量查询数据库"""
user_ids = [uid for uid, _ in batch]
# 批量查询
results = await self._batch_query_users(user_ids)
for (_, future), result in zip(batch, results):
if not future.done():
future.set_result(result)
async def _batch_query_users(self, user_ids):
"""模拟批量数据库查询"""
await asyncio.sleep(0.1)
return [{"user_id": uid, "name": f"User_{uid}"} for uid in user_ids]
batcher = RequestBatcher()
@app.get("/user/{user_id}")
async def get_user(user_id: int):
"""获取用户信息（请求合并版本）"""
# 检查缓存
cached = await cache.get(f"user:{user_id}")
if cached:
return cached
# 通过请求合并器获取
result = await batcher.add_request(user_id)
# 设置缓存
await cache.set(f"user:{user_id}", result)
return result
@app.post("/batch-process")
async def batch_process(items: list, background_tasks: BackgroundTasks):
"""批量处理接口"""
# 立即响应
background_tasks.add_task(process_items_in_background, items)
return {"status": "accepted", "total": len(items)}
async def process_items_in_background(items):
"""异步后台处理"""
semaphore = asyncio.Semaphore(10) # 限制并发数
async def process_one(item):
async with semaphore:
await asyncio.sleep(1) # 模拟处理
return item * 2
tasks = [process_one(item) for item in items]
results = await asyncio.gather(*tasks)
print(f"处理完成: {len(results)} 项")
# 启动
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
优化技巧总结
性能监控
import asyncio
import time
from functools import wraps
def async_timed(func):
@wraps(func)
async def wrapper(*args, **kwargs):
start = time.perf_counter()
try:
return await func(*args, **kwargs)
finally:
elapsed = time.perf_counter() - start
print(f"{func.__name__} 耗时: {elapsed:.3f}s")
return wrapper
@async_timed
async def concurrent_task():
pass
关键优化策略
- IO密集型：使用asyncio + aiohttp/aiofiles
- CPU密集型：使用多进程（ProcessPoolExecutor）
- 混合型：协程+进程组合
- 请求合并：减少数据库查询次数
- 缓存策略：使用缓存减少重复计算
- 连接池：复用网络连接
- 限流：使用信号量控制并发数
注意事项
- 避免不必要的内容切换
- 合理设置线程/进程池大小
- 使用批处理减少IO次数
- 注意内存使用，使用流式处理
这些案例涵盖了大多数并发优化的场景,你可以根据实际需求选择和组合使用。
