Python全系列大师课：深入异步编程底层，打造高并发数据管道

# 生成器可以暂停/恢复，这正是协程的原始模型
def gen_echo():
    while True:
        received = yield
        print(f'Received: {received}')

g = gen_echo()
next(g)          # 激活到第一个yield
g.send('Hello')  # 发送值并恢复

def sub_gen():
    yield 1
    yield 2

def main_gen():
    yield from sub_gen()   # 委托给子生成器
    yield 3

list(main_gen())  # [1,2,3]

import asyncio

class CustomAwaitable:
    def __await__(self):
        yield from asyncio.sleep(1).__await__()  # 代理
        return 42

async def main():
    result = await CustomAwaitable()
    print(result)

asyncio.run(main())

def _run_once(self):
    # 1. 获取过期定时器回调
    timeout = self._timer_handle_heap[0].when - self.time()
    # 2. 使用selectors轮询IO就绪
    event_list = self._selector.select(timeout)
    # 3. 处理IO事件（将回调加入就绪队列）
    for key, events in event_list:
        self._ready.append(key.data)
    # 4. 处理所有就绪回调（包括Timer和Task的step）
    for _ in range(len(self._ready)):
        handle = self._ready.popleft()
        handle._run()

pip install uvloop

import asyncio
import uvloop
import time

async def io_task(n):
    await asyncio.sleep(0.001)  # 模拟1ms IO
    return n

async def run(loop, n=10000):
    await asyncio.gather(*[io_task(i) for i in range(n)])

# 原生asyncio
start = time.perf_counter()
asyncio.run(run(asyncio.get_event_loop(), 10000))
print(f'原生: {time.perf_counter()-start:.3f}s')

# uvloop
uvloop.install()
start = time.perf_counter()
asyncio.run(run(asyncio.get_event_loop(), 10000))
print(f'uvloop: {time.perf_counter()-start:.3f}s')

import asyncio
import time

class AsyncRateLimiter:
    def __init__(self, rate, per=1.0):
        self.rate = rate          # 令牌数
        self.per = per            # 时间窗口
        self.tokens = rate
        self.updated_at = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self, tokens=1):
        async with self._lock:
            now = time.monotonic()
            elapsed = now - self.updated_at
            self.tokens = min(self.rate, self.tokens + elapsed * (self.rate / self.per))
            self.updated_at = now
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            deficit = tokens - self.tokens
            wait_time = deficit / (self.rate / self.per)
            await asyncio.sleep(wait_time)
            self.tokens = 0
            self.updated_at = time.monotonic()
            return True

class AsyncHttpClient:
    def __init__(self, max_retries=3, backoff=1.0):
        self.max_retries = max_retries
        self.backoff = backoff

    async def post(self, url, data):
        for attempt in range(self.max_retries):
            try:
                # 模拟网络IO
                await asyncio.sleep(0.005)  # 5ms响应
                if attempt == 0 and data['id'] % 5 == 0:
                    raise RuntimeError("模拟服务端错误")
                return {"status": 200, "body": "ok"}
            except Exception as e:
                if attempt == self.max_retries - 1:
                    raise
                wait = self.backoff * (2 ** attempt) + random.random()
                await asyncio.sleep(wait)

import asyncio
import random
from collections import deque
from dataclasses import dataclass, field

@dataclass
class Device:
    id: int
    queue: asyncio.Queue = field(default_factory=asyncio.Queue)
    async def produce(self, limiter, client):
        seq = 0
        while True:
            await limiter.acquire(tokens=2)  # 限流20tps? 实际根据rate调整
            data = {"id": self.id, "seq": seq, "timestamp": time.time()}
            try:
                resp = await client.post("http://fake", data)
                # 聚合统计（窗口滑动）
                await self.queue.put(data)
                seq += 1
            except Exception as e:
                print(f"Device {self.id} failed: {e}")
            await asyncio.sleep(0.05)  # 50ms周期

class Aggregator:
    def __init__(self, window_sec=1.0):
        self.window = window_sec
        self.buffer = deque()
        self.lock = asyncio.Lock()

    async def consume(self, device_queues, output_queue):
        while True:
            tasks = [q.get() for q in device_queues]
            done, pending = await asyncio.wait(tasks, timeout=0.1)
            for fut in done:
                data = fut.result()
                async with self.lock:
                    self.buffer.append(data)
                    # 清理过期数据（超过窗口）
                    cutoff = time.time() - self.window
                    while self.buffer and self.buffer[0]['timestamp'] < cutoff:
                        self.buffer.popleft()
                    # 计算每秒指标
                    if len(self.buffer) > 0:
                        avg_seq = sum(d['seq'] for d in self.buffer) / len(self.buffer)
                        output_queue.append(f"Window count: {len(self.buffer)}, avg seq: {avg_seq:.2f}")
            # 释放未完成的任务（避免无限等待）
            for fut in pending:
                fut.cancel()

async def main():
    limiter = AsyncRateLimiter(rate=200, per=1.0)  # 200 tokens/s
    client = AsyncHttpClient()
    devices = [Device(id=i) for i in range(100)]
    device_queues = [d.queue for d in devices]
    aggregator = Aggregator()
    output = asyncio.Queue()

    # 启动生产者和消费者
    producers = [asyncio.create_task(d.produce(limiter, client)) for d in devices]
    consumer = asyncio.create_task(aggregator.consume(device_queues, output))

    # 每秒打印聚合结果
    async def reporter():
        while True:
            await asyncio.sleep(1)
            if not output.empty():
                msgs = []
                while not output.empty():
                    msgs.append(await output.get())
                print(f"[REPORT] {', '.join(msgs)}")

    reporter_task = asyncio.create_task(reporter())

    try:
        await asyncio.sleep(5)  # 运行5秒演示
    finally:
        for p in producers:
            p.cancel()
        consumer.cancel()
        reporter_task.cancel()
        await asyncio.gather(*producers, consumer, reporter_task, return_exceptions=True)

if __name__ == "__main__":
    asyncio.run(main())

class ConnectionPool:
    async def __aenter__(self):
        self.conn = await self._create_conn()
        return self.conn
    async def __aexit__(self, *exc):
        await self._release(self.conn)

# 使用
async with pool.get_connection() as conn:
    await conn.execute(...)

loop = asyncio.get_running_loop()
tasks = asyncio.all_tasks(loop)
pending = [t for t in tasks if not t.done()]
print(f"Pending tasks: {len(pending)}")
# 打印堆栈（定位阻塞）
for t in pending[:5]:
    t.print_stack()