Python脚本如何优化异步任务执行效率

wen python案例 2026-06-27 244

本文目录导读：

使用高性能异步框架
合理控制并发数量
减少上下文切换开销
使用连接池和缓存
优化I/O操作
使用异步生成器
优化错误处理和超时
使用异步队列
性能监控和调优
完整优化示例
关键优化要点

我来详细介绍优化Python异步任务执行效率的关键方法：

使用高性能异步框架

# 选择更高效的框架
# 比较：asyncio vs uvloop vs trio
import asyncio
import uvloop
# 使用uvloop替代默认事件循环（性能提升2-4倍）
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

合理控制并发数量

import asyncio
from typing import List
async def process_task(task_id: int):
    # 单个任务处理逻辑
    await asyncio.sleep(0.1)
    return task_id
async def optimized_concurrent_processing(tasks: List[int]):
    """使用信号量控制并发"""
    semaphore = asyncio.Semaphore(10)  # 限制并发数为10
    async def limited_task(task_id):
        async with semaphore:
            return await process_task(task_id)
    # 创建所有任务
    coroutines = [limited_task(task) for task in tasks]
    return await asyncio.gather(*coroutines)
# 批量处理，避免一次性创建过多协程
async def batch_processing(tasks: List[int], batch_size: int = 100):
    results = []
    for i in range(0, len(tasks), batch_size):
        batch = tasks[i:i + batch_size]
        batch_results = await asyncio.gather(*[
            process_task(task) for task in batch
        ])
        results.extend(batch_results)
    return results

减少上下文切换开销

import asyncio
# 不好的实践：频繁的await
async def bad_practice():
    for i in range(1000):
        await asyncio.sleep(0)  # 每个迭代都切换上下文
# 好的实践：批量处理
async def good_practice():
    await asyncio.sleep(0)  # 一次性让出控制权
    for i in range(1000):
        # 密集计算，不进行await
        result = i * i

使用连接池和缓存

import asyncio
import aiohttp
from cachetools import TTLCache
class AsyncHttpClient:
    def __init__(self):
        # 使用连接池
        self.session = aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(
                limit=100,  # 连接池大小
                ttl_dns_cache=300,  # DNS缓存时间
                enable_cleanup_closed=True
            )
        )
        # 使用缓存
        self.cache = TTLCache(maxsize=1000, ttl=300)
    async def fetch(self, url: str):
        # 检查缓存
        if url in self.cache:
            return self.cache[url]
        async with self.session.get(url) as response:
            data = await response.text()
            self.cache[url] = data
            return data
    async def close(self):
        await self.session.close()

优化I/O操作

import asyncio
import aiofiles
# 异步文件操作
async def read_file_async(filepath: str):
    async with aiofiles.open(filepath, 'r') as f:
        content = await f.read()
    return content
# 批量I/O操作
async def batch_io_operations(filepaths: list):
    tasks = [read_file_async(path) for path in filepaths]
    return await asyncio.gather(*tasks)

使用异步生成器

import asyncio
from typing import AsyncGenerator
# 使用异步生成器减少内存占用
async def async_data_stream() -> AsyncGenerator[int, None]:
    for i in range(1000000):
        yield i
        # 每100个元素让出控制权
        if i % 100 == 0:
            await asyncio.sleep(0)
async def process_stream():
    async for item in async_data_stream():
        # 处理每个元素
        pass

优化错误处理和超时

import asyncio
async def robust_task(task_id: int, timeout: float = 5.0):
    try:
        # 使用asyncio.wait_for设置超时
        result = await asyncio.wait_for(
            process_task(task_id),
            timeout=timeout
        )
        return result
    except asyncio.TimeoutError:
        print(f"Task {task_id} timed out")
        return None
    except Exception as e:
        print(f"Task {task_id} failed: {e}")
        return None
# 优雅处理部分失败
async def handle_partial_failure(tasks: list):
    done, pending = await asyncio.wait(
        [robust_task(task) for task in tasks],
        return_when=asyncio.FIRST_EXCEPTION
    )
    # 取消超时的任务
    for task in pending:
        task.cancel()
    results = []
    for completed in done:
        try:
            result = completed.result()
            results.append(result)
        except:
            results.append(None)
    return results

使用异步队列

import asyncio
from asyncio import Queue
async def producer(queue: Queue, items: list):
    for item in items:
        await queue.put(item)
    await queue.put(None)  # 发送结束信号
async def consumer(queue: Queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        # 处理item
        await asyncio.sleep(0.1)
        queue.task_done()
async def queue_based_processing(items: list):
    queue = asyncio.Queue(maxsize=100)  # 限制队列大小
    # 启动多个consumer
    consumers = [asyncio.create_task(consumer(queue)) for _ in range(5)]
    # 启动producer
    await producer(queue, items)
    # 等待所有consumer完成
    await asyncio.gather(*consumers)

性能监控和调优

import asyncio
import time
from functools import wraps
def async_timed(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.time()
        try:
            return await func(*args, **kwargs)
        finally:
            elapsed = time.time() - start
            print(f"{func.__name__} took {elapsed:.3f} seconds")
    return wrapper
@async_timed
async def monitored_async_function():
    await asyncio.sleep(1)
# 使用asyncio.run的debug模式
asyncio.run(monitored_async_function(), debug=True)

完整优化示例

import asyncio
import uvloop
from typing import List, Any
import aiohttp
from cachetools import TTLCache
class OptimizedAsyncProcessor:
    def __init__(self, max_concurrent: int = 50, cache_size: int = 1000):
        # 使用uvloop
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.cache = TTLCache(maxsize=cache_size, ttl=300)
        self.session = aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(
                limit=max_concurrent,
                ttl_dns_cache=300
            )
        )
    async def process_item(self, item: Any) -> Any:
        """带信号量和缓存的异步处理"""
        async with self.semaphore:
            # 检查缓存
            if item in self.cache:
                return self.cache[item]
            # 处理逻辑
            result = await self._fetch_and_process(item)
            # 更新缓存
            self.cache[item] = result
            return result
    async def _fetch_and_process(self, item: Any) -> Any:
        """实际的获取和处理逻辑"""
        async with self.session.get(f"http://api.example.com/{item}") as response:
            data = await response.json()
        # 处理数据...
        return data
    async def process_batch(self, items: List[Any], batch_size: int = 100) -> List[Any]:
        """批量处理，避免内存溢出"""
        results = []
        for i in range(0, len(items), batch_size):
            batch = items[i:i + batch_size]
            batch_results = await asyncio.gather(*[
                self.process_item(item) for item in batch
            ])
            results.extend(batch_results)
        return results
    async def cleanup(self):
        """清理资源"""
        await self.session.close()
# 使用示例
async def main():
    processor = OptimizedAsyncProcessor()
    # 生成测试数据
    items = list(range(1000))
    # 处理数据
    results = await processor.process_batch(items)
    # 清理
    await processor.cleanup()
    print(f"Processed {len(results)} items")
if __name__ == "__main__":
    asyncio.run(main())