Python异步编程性能优化实战：从asyncio事件循环原理到uvloop百万并发工程体系

import asyncio
import selectors

async def inspect_event_loop():
    """深入理解事件循环的内部机制"""
    loop = asyncio.get_running_loop()
    
    # 底层选择器（在Linux上是epoll，macOS是kqueue）
    selector = loop._selector  # type: selectors.BaseSelector
    print(f"底层I/O多路复用: {type(selector).__name__}")
    print(f"事件循环时钟精度: {loop.clock_resolution:.6f}s")
    
    # 就绪队列中的任务数
    print(f"就绪回调数: {len(loop._ready)}")
    print(f"待调度回调数: {len(loop._scheduled)}")

asyncio.run(inspect_event_loop())

import asyncio

async def lifecycle_demo():
    """展示协程到Task的生命周期转化"""
    
    async def sample_work(name, delay):
        await asyncio.sleep(delay)
        return f"{name}完成"
    
    # 1. 普通协程 — 仅定义，不执行
    coro = sample_work("任务A", 0.1)
    print(f"协程对象: {type(coro)}")  # 
    
    # 2. Task — 协程被调度到事件循环中
    task = asyncio.create_task(coro)
    print(f"Task状态: done={task.done()}, cancelled={task.cancelled()}")
    
    # 3. result() 返回Future的结果
    result = await task
    print(f"结果: {result}, Task状态: done={task.done()}")

asyncio.run(lifecycle_demo())

import asyncio
import time
import aiohttp
import requests
from concurrent.futures import ThreadPoolExecutor
import uvloop

# 基准测试配置
CONCURRENT_REQUESTS = 10_000
TARGET_URL = "http://mock-api:8080/simulate?delay=50"

def benchmark_sync():
    """同步模式 — 需要大量线程"""
    start = time.perf_counter()
    with requests.Session() as session:
        with ThreadPoolExecutor(max_workers=500) as executor:
            results = list(executor.map(
                lambda: session.get(TARGET_URL), 
                range(CONCURRENT_REQUESTS)
            ))
    elapsed = time.perf_counter() - start
    print(f"同步(500线程): {elapsed:.2f}s | QPS: {CONCURRENT_REQUESTS/elapsed:.0f}")

async def benchmark_asyncio():
    """原生asyncio模式"""
    async def fetch(session, url):
        async with session.get(url) as resp:
            return resp.status
    
    start = time.perf_counter()
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, TARGET_URL) for _ in range(CONCURRENT_REQUESTS)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    elapsed = time.perf_counter() - start
    print(f"asyncio: {elapsed:.2f}s | QPS: {CONCURRENT_REQUESTS/elapsed:.0f}")

import aiohttp
from aiohttp import TCPConnector
import asyncio

async def optimized_http_client():
    """生产级异步HTTP客户端配置"""
    
    connector = TCPConnector(
        limit=200,                    # 总连接池上限
        limit_per_host=50,            # 单域名连接上限
        ttl_dns_cache=300,            # DNS缓存5分钟
        enable_cleanup_closed=True,   # 清理已关闭连接
        force_close=False,            # 保持连接复用
        enable_compression=True,      # 启用gzip压缩
    )
    
    timeout = aiohttp.ClientTimeout(
        total=30,                     # 总超时
        connect=5,                    # 连接超时
        sock_read=10,                 # 读超时
    )
    
    async with aiohttp.ClientSession(
        connector=connector,
        timeout=timeout,
        headers={"User-Agent": "MyApp/1.0"},
    ) as session:
        urls = [f"https://api.example.com/data/{i}" for i in range(1000)]
        semaphore = asyncio.Semaphore(50)  # 限流保护
        
        async def bounded_fetch(url):
            async with semaphore:
                async with session.get(url) as resp:
                    return await resp.json()
        
        results = await asyncio.gather(
            *[bounded_fetch(url) for url in urls],
            return_exceptions=True
        )
        
        successes = [r for r in results if not isinstance(r, Exception)]
        failures = [r for r in results if isinstance(r, Exception)]
        print(f"成功: {len(successes)}, 失败: {len(failures)}")

asyncio.run(optimized_http_client())

import asyncio
from collections import deque

class AdaptiveRateLimiter:
    """自适应限流控制器 — 根据响应时间动态调整并发度"""
    def __init__(self, initial_limit=50, min_limit=5, max_limit=200):
        self.limit = initial_limit
        self.min_limit = min_limit
        self.max_limit = max_limit
        self.semaphore = asyncio.Semaphore(initial_limit)
        self.latency_history = deque(maxlen=20)
        self.error_count = 0
    
    async def acquire(self):
        await self.semaphore.acquire()
    
    def release(self):
        self.semaphore.release()
    
    async def report_result(self, latency, error=False):
        """根据反馈调整限流"""
        if error:
            self.error_count += 1
            if self.error_count >= 3:
                new_limit = max(self.min_limit, self.limit // 2)
                self._adjust_limit(new_limit)
                self.error_count = 0
        else:
            self.error_count = 0
            self.latency_history.append(latency)
            if len(self.latency_history) == self.latency_history.maxlen:
                avg_latency = sum(self.latency_history) / len(self.latency_history)
                if avg_latency > 1.0:  # 平均延迟超1秒
                    new_limit = max(self.min_limit, int(self.limit * 0.8))
                    self._adjust_limit(new_limit)
                elif avg_latency < 0 xss=removed xss=removed xss=removed xss=removed xss=removed>