AI 技能AI 插件【免费下载链接】agentic-awesome-skillsAAS Core is the local, agent-first control plane for complete catalog discovery, agent-owned selection, stack validation, and planning, backed by 2,400 agentic skills. Includes CLI, local MCP, catalog, plugins, and Workbench.项目地址https://gitcode.com/gh_mirrors/an/agentic-awesome-skills点击查看免费下载导读本文是 agentic-awesome-skills 仓库中async-python-patterns技能SKILL.md的完整技术指南系统讲解如何使用 asyncio、async/await 与并发编程模式构建高性能、非阻塞的 Python 应用。文章以该技能的 implementation-playbook.md 为核心骨架完整覆盖 10 个基础与进阶模式、aiohttp 爬虫 / 异步数据库 / WebSocket 三类真实应用、性能最佳实践、常见陷阱与测试方法并结合仓库内真实异步实现如skills/junta-leiloeiros的并发抓取器提供源码级佐证。读完本文你将掌握事件循环、协程、Task、gather、Queue、Semaphore、Lock 等核心原语的正确用法并能在 FastAPI、aiohttp、Sanic 等场景中落地可并发、可取消、可观测的异步代码。一、技能定位何时使用 async-python-patterns在 agentic-awesome-skills 的 2400 技能体系中async-python-patterns是面向I/O 密集型异步 Python 应用的专项技能。它的元数据定义如下见 SKILL.mdname: async-python-patterns description: Comprehensive guidance for implementing asynchronous Python applications using asyncio, concurrent programming patterns, and async/await for building high-performance, non-blocking systems. risk: safe source: community date_added: 2026-02-271.1 适用场景Use this skill when根据技能定义以下场景应调用该技能构建异步 Web APIFastAPI、aiohttp、Sanic实现并发 I/O 操作数据库、文件、网络创建带并发请求的 Web 爬虫开发实时应用WebSocket 服务器、聊天系统同时处理多个相互独立的任务构建异步通信的微服务优化 I/O 密集型工作负载实现异步后台任务与队列。1.2 不适用场景Do not use this skill when工作负载是CPU 密集型且 I/O 极少此时应优先考虑多进程或计算库一个简单的同步脚本已经足够运行环境无法支持 asyncio / 事件循环如某些受限嵌入式环境或老版本 Python。1.3 技能使用流程Instructions技能要求使用者在编码前先完成四个前置动作澄清工作负载特征I/O 型还是 CPU 型、目标与运行时约束选择并发模式tasks、gather、queues、pools并明确取消cancellation规则补充超时timeouts、背压backpressure与结构化错误处理包含异步代码路径的测试与调试指引。当需要详细示例时技能明确要求打开resources/implementation-playbook.md仓库路径implementation-playbook.md下文所有模式均出自该文件。二、核心概念事件循环、协程、Task 与 Future在进入代码模式之前先建立对 asyncio 四大基础抽象的正确理解对应 playbook 的 Core Concepts 章节概念说明关键特征事件循环Event Loopasyncio 的心脏负责管理与调度异步任务单线程协作式多任务调度协程执行无阻塞地处理 I/O管理回调和 Future协程Coroutine用async def定义的、可暂停可恢复的函数通过await挂起/恢复不会自行执行必须被调度Task被调度到事件循环上并发运行的协程通过asyncio.create_task()创建独立调度执行Future表示异步操作最终结果的底层对象是 Task 的底层机制一般不需要直接操作协程的基础语法async def my_coroutine(): result await some_async_operation() return result2.1 异步上下文管理器与异步迭代器Async Context Managers支持async with语法用于资源的正确清理连接、会话、锁等对应后面的 Pattern 6Async Iterators支持async for语法用于遍历异步数据源分页 API、消息流等对应后面的 Pattern 7。2.2 快速开始import asyncio async def main(): print(Hello) await asyncio.sleep(1) print(World) # Python 3.7 asyncio.run(main())自 Python 3.7 起asyncio.run()是官方推荐的入口方式它负责创建事件循环、运行协程、并在结束后关闭循环避免了手动管理 loop 生命周期带来的资源泄漏风险。三、基础模式从顺序到并发Pattern 1–5Pattern 1基础 async/await这是最朴素的异步形态——单个协程内部通过await挂起等待 I/O 完成期间事件循环可以调度其他任务import asyncio async def fetch_data(url: str) - dict: Fetch data from URL asynchronously. await asyncio.sleep(1) # Simulate I/O return {url: url, data: result} async def main(): result await fetch_data(https://api.example.com) print(result) asyncio.run(main())Pattern 2用 gather() 并发执行当有多个相互独立的 I/O 任务时用asyncio.gather()一次性并发调度总耗时约等于最慢单个任务而非任务之和import asyncio from typing import List async def fetch_user(user_id: int) - dict: Fetch user data. await asyncio.sleep(0.5) return {id: user_id, name: fUser {user_id}} async def fetch_all_users(user_ids: List[int]) - List[dict]: Fetch multiple users concurrently. tasks [fetch_user(uid) for uid in user_ids] results await asyncio.gather(*tasks) return results async def main(): user_ids [1, 2, 3, 4, 5] users await fetch_all_users(user_ids) print(fFetched {len(users)} users) asyncio.run(main())仓库真实佐证skills/junta-leiloeiros/scripts/run_all.py巴西商业登记处抓取编排器正是这一模式的落地实现——它为 27 个州的抓取任务构建 task 列表后await asyncio.gather(*tasks)并发执行run_all.pysemaphore asyncio.Semaphore(concurrency) tasks [scrape_state(uf, semaphore) for uf in estados_alvo] results await asyncio.gather(*tasks)Pattern 3Task 的创建与管理asyncio.create_task()创建的任务会立即被调度到事件循环上并发运行主协程可以继续做其他工作之后再等待任务结果import asyncio async def background_task(name: str, delay: int): Long-running background task. print(f{name} started) await asyncio.sleep(delay) print(f{name} completed) return fResult from {name} async def main(): # Create tasks task1 asyncio.create_task(background_task(Task 1, 2)) task2 asyncio.create_task(background_task(Task 2, 1)) # Do other work print(Main: doing other work) await asyncio.sleep(0.5) # Wait for tasks result1 await task1 result2 await task2 print(fResults: {result1}, {result2}) asyncio.run(main())要点create_task 返回的 Task 对象本身可await如果创建后不等待也不取消会在程序退出时收到 Task was destroyed but it is pending 告警因此必须管理好每个 Task 的生命周期。Pattern 4异步代码的错误处理asyncio.gather(..., return_exceptionsTrue)让异常以返回值形式收集到结果列表中从而在聚合层面统一分流成功与失败import asyncio from typing import List, Optional async def risky_operation(item_id: int) - dict: Operation that might fail. await asyncio.sleep(0.1) if item_id % 3 0: raise ValueError(fItem {item_id} failed) return {id: item_id, status: success} async def safe_operation(item_id: int) - Optional[dict]: Wrapper with error handling. try: return await risky_operation(item_id) except ValueError as e: print(fError: {e}) return None async def process_items(item_ids: List[int]): Process multiple items with error handling. tasks [safe_operation(iid) for iid in item_ids] results await asyncio.gather(*tasks, return_exceptionsTrue) # Filter out failures successful [r for r in results if r is not None and not isinstance(r, Exception)] failed [r for r in results if isinstance(r, Exception)] print(fSuccess: {len(successful)}, Failed: {len(failed)}) return successful asyncio.run(process_items([1, 2, 3, 4, 5, 6]))仓库真实佐证run_all.py中每个州的抓取都包在try/except Exception中任何单个州的失败都会被捕获并写入带status: ERRO的结果字典不会拖垮整个批次run_all.py——这正是结构化错误处理在真实编排器中的体现。Pattern 5超时处理用asyncio.wait_for()为易挂起的操作设定截止时间超时抛出asyncio.TimeoutErrorimport asyncio async def slow_operation(delay: int) - str: Operation that takes time. await asyncio.sleep(delay) return fCompleted after {delay}s async def with_timeout(): Execute operation with timeout. try: result await asyncio.wait_for(slow_operation(5), timeout2.0) print(result) except asyncio.TimeoutError: print(Operation timed out) asyncio.run(with_timeout())四、进阶模式资源控制与同步Pattern 6–10Pattern 6异步上下文管理器自定义__aenter__/__aexit__可以让连接、会话等资源在async with块退出时自动、异步地完成清理import asyncio from typing import Optional class AsyncDatabaseConnection: Async database connection context manager. def __init__(self, dsn: str): self.dsn dsn self.connection: Optional[object] None async def __aenter__(self): print(Opening connection) await asyncio.sleep(0.1) # Simulate connection self.connection {dsn: self.dsn, connected: True} return self.connection async def __aexit__(self, exc_type, exc_val, exc_tb): print(Closing connection) await asyncio.sleep(0.1) # Simulate cleanup self.connection None async def query_database(): Use async context manager. async with AsyncDatabaseConnection(postgresql://localhost) as conn: print(fUsing connection: {conn}) await asyncio.sleep(0.2) # Simulate query return {rows: 10} asyncio.run(query_database())Pattern 7异步迭代器与生成器用async for消费async defyield定义的异步生成器适合分页拉取等按需生产场景import asyncio from typing import AsyncIterator async def async_range(start: int, end: int, delay: float 0.1) - AsyncIterator[int]: Async generator that yields numbers with delay. for i in range(start, end): await asyncio.sleep(delay) yield i async def fetch_pages(url: str, max_pages: int) - AsyncIterator[dict]: Fetch paginated data asynchronously. for page in range(1, max_pages 1): await asyncio.sleep(0.2) # Simulate API call yield { page: page, url: f{url}?page{page}, data: [fitem_{page}_{i} for i in range(5)] } async def consume_async_iterator(): Consume async iterator. async for number in async_range(1, 5): print(fNumber: {number}) print(\nFetching pages:) async for page_data in fetch_pages(https://api.example.com/items, 3): print(fPage {page_data[page]}: {len(page_data[data])} items) asyncio.run(consume_async_iterator())Pattern 8生产者-消费者模式异步队列asyncio.Queue天然支持生产/消费解耦生产者put数据消费者get处理None作为终止信号queue.join()等待队列清空最后显式取消消费者 Taskimport asyncio from asyncio import Queue from typing import Optional async def producer(queue: Queue, producer_id: int, num_items: int): Produce items and put them in queue. for i in range(num_items): item fItem-{producer_id}-{i} await queue.put(item) print(fProducer {producer_id} produced: {item}) await asyncio.sleep(0.1) await queue.put(None) # Signal completion async def consumer(queue: Queue, consumer_id: int): Consume items from queue. while True: item await queue.get() if item is None: queue.task_done() break print(fConsumer {consumer_id} processing: {item}) await asyncio.sleep(0.2) # Simulate work queue.task_done() async def producer_consumer_example(): Run producer-consumer pattern. queue Queue(maxsize10) # Create tasks producers [ asyncio.create_task(producer(queue, i, 5)) for i in range(2) ] consumers [ asyncio.create_task(consumer(queue, i)) for i in range(3) ] # Wait for producers await asyncio.gather(*producers) # Wait for queue to be empty await queue.join() # Cancel consumers for c in consumers: c.cancel() asyncio.run(producer_consumer_example())注意Queue(maxsize10)本身就提供了背压backpressure当队列满时put会挂起等待消费者取走元素防止生产速度无限超过消费速度——这正是技能 Instructions 中要求补充背压的典型实现。Pattern 9用 Semaphore 实现限流Rate Limiting当并发任务数量可能压垮下游服务时用asyncio.Semaphore控制同时进行的任务上限import asyncio from typing import List async def api_call(url: str, semaphore: asyncio.Semaphore) - dict: Make API call with rate limiting. async with semaphore: print(fCalling {url}) await asyncio.sleep(0.5) # Simulate API call return {url: url, status: 200} async def rate_limited_requests(urls: List[str], max_concurrent: int 5): Make multiple requests with rate limiting. semaphore asyncio.Semaphore(max_concurrent) tasks [api_call(url, semaphore) for url in urls] results await asyncio.gather(*tasks) return results async def main(): urls [fhttps://api.example.com/item/{i} for i in range(20)] results await rate_limited_requests(urls, max_concurrent3) print(fCompleted {len(results)} requests) asyncio.run(main())仓库真实佐证这正是skills/junta-leiloeiros/scripts/run_all.py的核心并发策略——它把--concurrency默认 5作为asyncio.Semaphore的初始值每个州的scrape_state进入时async with semaphore从而在 27 个抓取任务之间精确控制并行度run_all.py避免对目标站点造成过大压力。Pattern 10异步锁与同步单线程事件循环内虽然不会发生真正的数据竞争但多个协程交错执行时仍可能破坏读-改-写的原子性。用asyncio.Lock保护临界区import asyncio class AsyncCounter: Thread-safe async counter. def __init__(self): self.value 0 self.lock asyncio.Lock() async def increment(self): Safely increment counter. async with self.lock: current self.value await asyncio.sleep(0.01) # Simulate work self.value current 1 async def get_value(self) - int: Get current value. async with self.lock: return self.value async def worker(counter: AsyncCounter, worker_id: int): Worker that increments counter. for _ in range(10): await counter.increment() print(fWorker {worker_id} incremented) async def test_counter(): Test concurrent counter. counter AsyncCounter() workers [asyncio.create_task(worker(counter, i)) for i in range(5)] await asyncio.gather(*workers) final_value await counter.get_value() print(fFinal counter value: {final_value}) asyncio.run(test_counter())如果去掉锁5 个 worker 各自执行 10 次读取→睡眠→写回后value将远小于 50加锁后最终值稳定为 50。五、真实应用把模式组装成系统5.1 基于 aiohttp 的并发 Web 爬虫将 Session 复用、gather 并发、超时与异常兜底组合成最小可用爬虫import asyncio import aiohttp from typing import List, Dict async def fetch_url(session: aiohttp.ClientSession, url: str) - Dict: Fetch single URL. try: async with session.get(url, timeoutaiohttp.ClientTimeout(total10)) as response: text await response.text() return { url: url, status: response.status, length: len(text) } except Exception as e: return {url: url, error: str(e)} async def scrape_urls(urls: List[str]) - List[Dict]: Scrape multiple URLs concurrently. async with aiohttp.ClientSession() as session: tasks [fetch_url(session, url) for url in urls] results await asyncio.gather(*tasks) return results async def main(): urls [ https://httpbin.org/delay/1, https://httpbin.org/delay/2, https://httpbin.org/status/404, ] results await scrape_urls(urls) for result in results: print(result) asyncio.run(main())其中aiohttp.ClientTimeout(total10)在请求层实现了 Pattern 5 的超时语义单个 URL 的异常被收敛为结果字典中的error字段Pattern 4 的错误分流思路。5.2 异步数据库操作并发聚合查询把同一用户的多条查询并发执行显著降低接口时延import asyncio from typing import List, Optional # Simulated async database client class AsyncDB: Simulated async database. async def execute(self, query: str) - List[dict]: Execute query. await asyncio.sleep(0.1) return [{id: 1, name: Example}] async def fetch_one(self, query: str) - Optional[dict]: Fetch single row. await asyncio.sleep(0.1) return {id: 1, name: Example} async def get_user_data(db: AsyncDB, user_id: int) - dict: Fetch user and related data concurrently. user_task db.fetch_one(fSELECT * FROM users WHERE id {user_id}) orders_task db.execute(fSELECT * FROM orders WHERE user_id {user_id}) profile_task db.fetch_one(fSELECT * FROM profiles WHERE user_id {user_id}) user, orders, profile await asyncio.gather(user_task, orders_task, profile_task) return { user: user, orders: orders, profile: profile } async def main(): db AsyncDB() user_data await get_user_data(db, 1) print(user_data) asyncio.run(main())5.3 WebSocket 服务器注册/广播/消息迭代WebSocket 天然适合异步模型——每个连接都是一个持续等待消息的协程import asyncio from typing import Set # Simulated WebSocket connection class WebSocket: Simulated WebSocket. def __init__(self, client_id: str): self.client_id client_id async def send(self, message: str): Send message. print(fSending to {self.client_id}: {message}) await asyncio.sleep(0.01) async def recv(self) - str: Receive message. await asyncio.sleep(1) return fMessage from {self.client_id} class WebSocketServer: Simple WebSocket server. def __init__(self): self.clients: Set[WebSocket] set() async def register(self, websocket: WebSocket): Register new client. self.clients.add(websocket) print(fClient {websocket.client_id} connected) async def unregister(self, websocket: WebSocket): Unregister client. self.clients.remove(websocket) print(fClient {websocket.client_id} disconnected) async def broadcast(self, message: str): Broadcast message to all clients. if self.clients: tasks [client.send(message) for client in self.clients] await asyncio.gather(*tasks) async def handle_client(self, websocket: WebSocket): Handle individual client connection. await self.register(websocket) try: async for message in self.message_iterator(websocket): await self.broadcast(f{websocket.client_id}: {message}) finally: await self.unregister(websocket) async def message_iterator(self, websocket: WebSocket): Iterate over messages from client. for _ in range(3): # Simulate 3 messages yield await websocket.recv()这个实现把 Pattern 7async 迭代器接收消息、Pattern 2gather 广播与try/finally清理保证断开时注销连接组合成了一个完整的小型聊天服务骨架。六、性能最佳实践6.1 使用连接池为 aiohttp 显式配置TCPConnector用limit控制全局并发连接数、limit_per_host控制单主机并发上限import asyncio import aiohttp async def with_connection_pool(): Use connection pool for efficiency. connector aiohttp.TCPConnector(limit100, limit_per_host10) async with aiohttp.ClientSession(connectorconnector) as session: tasks [session.get(fhttps://api.example.com/item/{i}) for i in range(50)] responses await asyncio.gather(*tasks) return responses6.2 批量操作对海量任务分批gather避免一次性创建数千个 Task 造成事件循环过载同时保留每批之间的可观测输出async def batch_process(items: List[str], batch_size: int 10): Process items in batches. for i in range(0, len(items), batch_size): batch items[i:i batch_size] tasks [process_item(item) for item in batch] await asyncio.gather(*tasks) print(fProcessed batch {i // batch_size 1}) async def process_item(item: str): Process single item. await asyncio.sleep(0.1) return fProcessed: {item}6.3 避免阻塞操作run_in_executor阻塞调用time.sleep、同步库、CPU 密集段会冻结整个事件循环必须移入线程池/进程池执行import asyncio import concurrent.futures from typing import Any def blocking_operation(data: Any) - Any: CPU-intensive blocking operation. import time time.sleep(1) return data * 2 async def run_in_executor(data: Any) - Any: Run blocking operation in thread pool. loop asyncio.get_event_loop() with concurrent.futures.ThreadPoolExecutor() as pool: result await loop.run_in_executor(pool, blocking_operation, data) return result async def main(): results await asyncio.gather(*[run_in_executor(i) for i in range(5)]) print(results) asyncio.run(main())注意对于真正的 CPU 密集型工作事件循环本就并非最佳工具——这正对应 SKILL.md 中Do not use this skill when的第一条边界。七、常见陷阱Common Pitfalls7.1 忘记 await调用 async 函数不await只会拿到一个未执行的协程对象# Wrong - returns coroutine object, doesnt execute result async_function() # Correct result await async_function()7.2 阻塞事件循环在协程内使用time.sleep会阻塞整个事件循环导致所有并发任务停滞# Wrong - blocks event loop import time async def bad(): time.sleep(1) # Blocks! # Correct async def good(): await asyncio.sleep(1) # Non-blocking7.3 不处理取消Cancellation长时间运行的 Task 被取消时应捕获asyncio.CancelledError完成清理后重新抛出确保取消语义正确传播async def cancelable_task(): Task that handles cancellation. try: while True: await asyncio.sleep(1) print(Working...) except asyncio.CancelledError: print(Task cancelled, cleaning up...) # Perform cleanup raise # Re-raise to propagate cancellation7.4 混用同步与异步代码async 函数内不能直接await语法错误同步入口应通过asyncio.run()桥接# Wrong - cant call async from sync directly def sync_function(): result await async_function() # SyntaxError! # Correct def sync_function(): result asyncio.run(async_function())八、测试异步代码配合 pytest-asyncio用pytest.mark.asyncio标记异步测试用例并可用pytest.raises断言超时等异常路径import asyncio import pytest # Using pytest-asyncio pytest.mark.asyncio async def test_async_function(): Test async function. result await fetch_data(https://api.example.com) assert result is not None pytest.mark.asyncio async def test_with_timeout(): Test with timeout. with pytest.raises(asyncio.TimeoutError): await asyncio.wait_for(slow_operation(5), timeout1.0)九、最佳实践速查表playbook 在最后给出了 10 条可直接落地的经验总结implementation-playbook.md用asyncio.run()作为入口Python 3.7总是await协程使其真正执行多任务并发优先用gather()用 try/except 实现结构化错误处理用超时防止操作无限挂起连接池化以提升性能异步代码中避免阻塞操作用 Semaphore 做限流正确处任务取消Cancellation用 pytest-asyncio 测试异步代码。十、仓库中的更多异步实战参考除本技能外agentic-awesome-skills 仓库中还有多个以 asyncio 为核心的真实实现可供对照学习run_all.pySemaphore 限流 gather 并发的多目标抓取编排器--concurrency参数直接控制并行度voice-ai-engine-development语音 Agent 引擎示例如 complete_voice_engine.py体现异步在实时流式场景的应用instagram脚本体系大量使用 asyncio 编排 API 调用如 publish.pyskill-sentinel 与 mcp-builder分析/评估类工具中的异步调度。结语async-python-patterns技能的价值在于它不是零散代码片段而是一套先澄清负载特征 → 再选择并发原语 → 后补超时/背压/错误处理 → 最后测试验证的完整方法论。事件循环与协程提供了单线程内的高并发骨架gather/Queue/Semaphore/Lock 分别解决聚合、生产消费、限流与同步问题而超时、取消与连接池则是让系统在生产环境保持健壮的关键护栏。无论是构建 FastAPI 异步 API、aiohttp 爬虫还是 WebSocket 实时服务都可以从这套模式库中直接取材组合仓库中run_all.py等真实实现则证明了这套模式在规模化 I/O 编排场景中的落地价值。赞分享AI 技能AI 插件【免费下载链接】agentic-awesome-skillsAAS Core is the local, agent-first control plane for complete catalog discovery, agent-owned selection, stack validation, and planning, backed by 2,400 agentic skills. Includes CLI, local MCP, catalog, plugins, and Workbench.项目地址https://gitcode.com/gh_mirrors/an/agentic-awesome-skills点击查看免费下载相关推荐Async Python Patterns 实战指南基于 asyncio 的高性能非阻塞应用开发手册Async Python Patterns 实战指南基于 asyncio 的高性能非阻塞应用开发手册 导读 本文以 agentic awesome skillAI 技能AI 插件redis-py异步编程指南利用asyncio实现高性能非阻塞操作redis py异步编程指南利用asyncio实现高性能非阻塞操作 你是否在处理高并发Redis操作时遇到过性能瓶颈传统同步客户端会阻塞主线程导致应用响应后端数据库客户端缓存3 分钟快速上手 bypass-paywalls-clean-filters新手安装与订阅完整教程3 分钟快速上手 bypass paywalls clean filters新手安装与订阅完整教程 你是否遇到过这样的情况好不容易打开一篇想看的新闻却被半开发工具创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考