Python并发优化案例如何优化并发逻辑

wen python案例 2026-06-28 277

本文目录导读：

基础案例：文件批量处理优化
进阶案例：爬虫数据采集优化
高性能案例：数据处理管道
实际案例：Web API服务优化
优化技巧总结

我来分享几个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