Python异步性能破壁：从事件循环原理到亿级并发上下文管理

import asyncio
import uvloop
from asyncio import AbstractEventLoop, Future

# 激活uvloop替代默认事件循环
uvloop.install()

class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
    """自定义事件循环策略：强制使用uvloop并注入性能监控钩子"""
    def new_event_loop(self) -> AbstractEventLoop:
        loop = uvloop.new_event_loop()
        # 开启调试模式，监控慢回调
        loop.set_debug(True)
        loop.slow_callback_duration = 0.1  # 超过100ms触发警告
        return loop

# 全局替换策略
asyncio.set_event_loop_policy(CustomEventLoopPolicy())

import time
from typing import AsyncContextManager, Optional, Dict
from collections import deque

class AsyncRateLimiter:
    """基于令牌桶算法的异步限流器，支持动态水位调节"""
    def __init__(self, rate: float, capacity: int):
        self.rate = rate  # 每秒令牌生成速率
        self.capacity = capacity  # 桶容量
        self._tokens = capacity
        self._last_refill = time.monotonic()
        self._lock = asyncio.Lock()
        self._waiters: deque[Future] = deque()

    async def acquire(self) -> bool:
        """非阻塞获取令牌，若失败则挂起当前协程"""
        async with self._lock:
            self._refill()
            if self._tokens >= 1:
                self._tokens -= 1
                return True
            # 创建等待Future并注册到队列
            fut = asyncio.get_running_loop().create_future()
            self._waiters.append(fut)
        
        try:
            await fut
            return True
        except asyncio.CancelledError:
            # 处理取消状态，防止协程泄漏
            async with self._lock:
                self._waiters.remove(fut)
            raise

    def _refill(self) -> None:
        now = time.monotonic()
        elapsed = now - self._last_refill
        new_tokens = elapsed * self.rate
        self._tokens = min(self.capacity, self._tokens + new_tokens)
        self._last_refill = now

    async def __aenter__(self):
        await self.acquire()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 释放时唤醒等待队列中的协程
        async with self._lock:
            self._refill()
            while self._waiters and self._tokens >= 1:
                waiter = self._waiters.popleft()
                if not waiter.done():
                    waiter.set_result(True)
                    self._tokens -= 1

# 使用示例：每秒处理5000个请求，峰值缓冲10000
limiter = AsyncRateLimiter(rate=5000, capacity=10000)

async def burst_processor(data: bytes):
    async with limiter:
        # 执行实际IO操作（如写入Kafka）
        await asyncio.sleep(0.001)  # 模拟IO

from typing import AsyncGenerator, Any
import asyncio

class StreamPipeline:
    def __init__(self, maxsize: int = 1000):
        self.queue: asyncio.Queue = asyncio.Queue(maxsize=maxsize)

    async def producer(self, data_source: AsyncGenerator[bytes, None]) -> None:
        """数据生产者，遇到队列满则自动阻塞"""
        async for chunk in data_source:
            await self.queue.put(chunk)  # 隐式反压

    async def consumer(self) -> AsyncGenerator[bytes, None]:
        """消费者异步生成器，支持async for迭代"""
        while True:
            item = await self.queue.get()
            if item is None:  # 毒丸信号
                break
            yield item
            self.queue.task_done()

    async def run(self, source: AsyncGenerator[bytes, None]) -> None:
        """启动流水线"""
        producer_task = asyncio.create_task(self.producer(source))
        consumer_task = asyncio.create_task(self.consume())
        await producer_task
        await self.queue.join()  # 等待所有任务完成
        consumer_task.cancel()   # 优雅关闭

import contextvars
from contextvars import ContextVar

trace_id_var: ContextVar[str] = ContextVar('trace_id', default='unknown')

class TraceContext:
    """异步上下文自动埋点管理器"""
    def __init__(self, trace_id: str):
        self.trace_id = trace_id
        self._token: Optional[contextvars.Token] = None

    async def __aenter__(self):
        self._token = trace_id_var.set(self.trace_id)
        # 返回绑定后的上下文对象，便于链式调用
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self._token:
            trace_id_var.reset(self._token)

# 深度结合：在限流器内部注入TraceID
class TracedRateLimiter(AsyncRateLimiter):
    async def __aenter__(self):
        # 从当前上下文获取trace_id并记录到日志
        current_trace = trace_id_var.get()
        print(f"[Trace: {current_trace}] Acquiring rate limit")
        return await super().__aenter__()

import time
from concurrent.futures import ThreadPoolExecutor

async def benchmark_async(rate: int, total: int):
    limiter = AsyncRateLimiter(rate, rate * 2)
    start = time.perf_counter()
    
    async def worker(i):
        async with limiter:
            pass  # 纯调度开销
    
    await asyncio.gather(*[worker(i) for i in range(total)])
    return time.perf_counter() - start

def benchmark_threadpool(rate: int, total: int):
    # 使用线程池模拟同步限流（糟糕实践）
    with ThreadPoolExecutor(max_workers=100) as executor:
        # ... 省略同步锁实现
        pass

if __name__ == "__main__":
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    elapsed = loop.run_until_complete(benchmark_async(rate=5000, total=100000))
    print(f"处理10万请求耗时: {elapsed:.2f}s, QPS: {100000/elapsed:.0f}")