Python协程与异步模式进阶

Python协程与异步模式进阶一、协程的本质协程是可以暂停和恢复执行的函数。Python中协程经历了三代演进- 基于生成器的协程Python 2.5已废弃- yield from协程Python 3.3- async/await原生协程Python 3.5推荐import asyncioimport inspectasync def my_coroutine():await asyncio.sleep(1)return 完成# 协程对象coro my_coroutine()print(type(coro)) #print(inspect.iscoroutine(coro)) # True# 必须通过事件循环或await驱动result asyncio.run(coro)二、事件循环深入事件循环是asyncio的核心负责调度协程、处理IO事件和回调。import asyncioasync def main():loop asyncio.get_running_loop()# 调度回调loop.call_soon(lambda: print(立即执行))loop.call_later(1.0, lambda: print(1秒后执行))# 在线程池中运行阻塞代码result await loop.run_in_executor(None, blocking_function)# 创建Futurefuture loop.create_future()loop.call_later(1.0, future.set_result, 延迟结果)result await futureprint(result)# 自定义事件循环策略class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy):def new_event_loop(self):loop super().new_event_loop()loop.set_debug(True)return loopasyncio.set_event_loop_policy(CustomEventLoopPolicy())三、结构化并发Python 3.11引入TaskGroup提供结构化并发async def fetch_data(url, delay):await asyncio.sleep(delay)if error in url:raise ValueError(f获取 {url} 失败)return f{url} 的数据# TaskGroup确保所有任务完成或全部取消async def fetch_all():results {}async with asyncio.TaskGroup() as tg:task1 tg.create_task(fetch_data(api/users, 1))task2 tg.create_task(fetch_data(api/posts, 2))task3 tg.create_task(fetch_data(api/comments, 0.5))# 所有任务成功完成后才到这里results[users] task1.result()results[posts] task2.result()results[comments] task3.result()return results# 错误处理async def fetch_with_errors():try:async with asyncio.TaskGroup() as tg:tg.create_task(fetch_data(api/users, 1))tg.create_task(fetch_data(api/error, 0.5)) # 会失败except* ValueError as eg:# except* 处理ExceptionGroupfor exc in eg.exceptions:print(f错误: {exc})四、异步设计模式4.1 扇出/扇入模式async def fan_out_fan_in(urls, max_concurrent10):限制并发数的批量请求semaphore asyncio.Semaphore(max_concurrent)results []async def limited_fetch(url):async with semaphore:return await fetch_data(url, 0.5)async with asyncio.TaskGroup() as tg:tasks [tg.create_task(limited_fetch(url)) for url in urls]return [task.result() for task in tasks]4.2 生产者-消费者模式async def producer(queue, items):for item in items:await asyncio.sleep(0.1) # 模拟生产耗时await queue.put(item)print(f生产: {item})# 发送停止信号await queue.put(None)async def consumer(queue, name):while True:item await queue.get()if item is None:await queue.put(None) # 传递停止信号给其他消费者breakawait asyncio.sleep(0.2) # 模拟消费耗时print(f{name} 消费: {item})queue.task_done()async def pipeline():queue asyncio.Queue(maxsize5)items list(range(20))async with asyncio.TaskGroup() as tg:tg.create_task(producer(queue, items))tg.create_task(consumer(queue, 消费者A))tg.create_task(consumer(queue, 消费者B))tg.create_task(consumer(queue, 消费者C))4.3 超时与重试async def with_timeout(coro, timeout, defaultNone):带超时的协程执行try:return await asyncio.wait_for(coro, timeouttimeout)except asyncio.TimeoutError:return defaultasync def with_retry(coro_factory, max_retries3, backoff1.0, exceptions(Exception,)):带重试的协程执行for attempt in range(max_retries):try:return await coro_factory()except exceptions as e:if attempt max_retries - 1:raisewait backoff * (2 ** attempt)print(f重试 {attempt 1}/{max_retries}等待 {wait}s: {e})await asyncio.sleep(wait)# 使用result await with_retry(lambda: fetch_data(api/unstable, 1),max_retries3,exceptions(ConnectionError, TimeoutError))4.4 异步上下文管理器池class AsyncPool:异步资源池def __init__(self, factory, max_size10):self._factory factoryself._pool asyncio.Queue(maxsizemax_size)self._size 0self._max_size max_sizeasync def acquire(self):try:return self._pool.get_nowait()except asyncio.QueueEmpty:if self._size self._max_size:self._size 1return await self._factory()return await self._pool.get()async def release(self, resource):await self._pool.put(resource)async def __aenter__(self):self._resource await self.acquire()return self._resourceasync def __aexit__(self, *args):await self.release(self._resource)五、异步迭代器模式class AsyncBatcher:异步批处理迭代器def __init__(self, source, batch_size10):self.source sourceself.batch_size batch_sizedef __aiter__(self):return self._iterate()async def _iterate(self):batch []async for item in self.source:batch.append(item)if len(batch) self.batch_size:yield batchbatch []if batch:yield batchclass AsyncMapper:异步映射迭代器def __init__(self, source, func, concurrency5):self.source sourceself.func funcself.concurrency concurrencydef __aiter__(self):return self._iterate()async def _iterate(self):semaphore asyncio.Semaphore(self.concurrency)async def process(item):async with semaphore:return await self.func(item)async for batch in AsyncBatcher(self.source, self.concurrency):tasks [asyncio.create_task(process(item)) for item in batch]for task in tasks:yield await task# 使用async def transform(item):await asyncio.sleep(0.1)return item * 2async def main():source async_range(100)mapper AsyncMapper(source, transform, concurrency10)async for result in mapper:print(result)六、异步信号与事件class AsyncEventBus:异步事件总线def __init__(self):self._handlers {}def on(self, event_name):def decorator(func):if event_name not in self._handlers:self._handlers[event_name] []self._handlers[event_name].append(func)return funcreturn decoratorasync def emit(self, event_name, *args, **kwargs):handlers self._handlers.get(event_name, [])tasks [asyncio.create_task(handler(*args, **kwargs))for handler in handlers]if tasks:await asyncio.gather(*tasks, return_exceptionsTrue)bus AsyncEventBus()bus.on(user_created)async def send_welcome_email(user):await asyncio.sleep(0.5)print(f发送欢迎邮件给 {user[name]})bus.on(user_created)async def init_user_settings(user):await asyncio.sleep(0.3)print(f初始化 {user[name]} 的设置)async def create_user(name, email):user {name: name, email: email}await bus.emit(user_created, user)return user七、与同步代码集成import asynciofrom concurrent.futures import ThreadPoolExecutor# 在异步代码中调用同步函数async def async_wrapper():loop asyncio.get_running_loop()# 使用默认线程池result await loop.run_in_executor(None, sync_blocking_function)# 使用自定义线程池with ThreadPoolExecutor(max_workers4) as pool:result await loop.run_in_executor(pool, cpu_intensive_function)return result# 在同步代码中调用异步函数def sync_function():# 方式1asyncio.run创建新事件循环result asyncio.run(async_function())# 方式2在已有循环中如Jupyter# import nest_asyncio# nest_asyncio.apply()# loop asyncio.get_event_loop()# result loop.run_until_complete(async_function())# 混合同步/异步的类class HybridService:def __init__(self):self._executor ThreadPoolExecutor(max_workers4)async def async_method(self):# 异步操作await asyncio.sleep(1)return async resultdef sync_method(self):# 同步操作return sync resultasync def call_sync_from_async(self):loop asyncio.get_running_loop()return await loop.run_in_executor(self._executor, self.sync_method)八、调试异步代码import asyncioimport logging# 启用调试模式asyncio.run(main(), debugTrue)# 或通过环境变量# PYTHONASYNCIODEBUG1 python script.py# 追踪慢回调logging.basicConfig(levellogging.DEBUG)# asyncio会警告执行超过100ms的回调# 检测未await的协程import warningswarnings.filterwarnings(error, categoryRuntimeWarning)# 任务异常追踪async def safe_task(coro, name):try:return await coroexcept Exception as e:logging.error(f任务 {name} 失败: {e}, exc_infoTrue)raise九、性能优化# 1. 避免在异步代码中使用阻塞调用# 错误async def bad():time.sleep(1) # 阻塞整个事件循环# 正确async def good():await asyncio.sleep(1)# 2. 批量操作减少await次数# 慢async def slow():for url in urls:await fetch(url)# 快async def fast():await asyncio.gather(*[fetch(url) for url in urls])# 3. 使用uvloop提升性能Linux/Mac# pip install uvloopimport uvloopasyncio.set_event_loop_policy(uvloop.EventLoopPolicy())# 4. 连接池复用# 不要每次请求都创建新连接async def with_pool():async with aiohttp.ClientSession() as session:# 复用session中的连接池for url in urls:async with session.get(url) as resp:pass十、总结异步编程进阶要点1. 理解事件循环的调度机制2. 使用TaskGroup实现结构化并发3. 用Semaphore控制并发度4. 异步迭代器适合流式数据处理5. run_in_executor桥接同步和异步代码6. 生产环境使用uvloop提升性能7. 开启debug模式排查问题8. 避免在异步代码中使用任何阻塞调用