agentic-awesome-skills 异步 Python 模式实战：基于 asyncio 构建高性能非阻塞系统

• AI 技能
• AI 插件

【免费下载链接】agentic-awesome-skills

AAS 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.md）：

name: 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-27"

1.1 适用场景（Use this skill when）

根据技能定义，以下场景应调用该技能：

• 构建异步 Web API（FastAPI、aiohttp、Sanic）；
• 实现并发 I/O 操作（数据库、文件、网络）；
• 创建带并发请求的 Web 爬虫；
• 开发实时应用（WebSocket 服务器、聊天系统）；
• 同时处理多个相互独立的任务；
• 构建异步通信的微服务；
• 优化 I/O 密集型工作负载；
• 实现异步后台任务与队列。

1.2 不适用场景（Do not use this skill when）

• 工作负载是CPU 密集型且 I/O 极少（此时应优先考虑多进程或计算库）；
• 一个简单的同步脚本已经足够；
• 运行环境无法支持 asyncio / 事件循环（如某些受限嵌入式环境或老版本 Python）。

1.3 技能使用流程（Instructions）

技能要求使用者在编码前先完成四个前置动作：

1. 澄清工作负载特征（I/O 型还是 CPU 型）、目标与运行时约束；
2. 选择并发模式（tasks、gather、queues、pools）并明确取消（cancellation）规则；
3. 补充超时（timeouts）、背压（backpressure）与结构化错误处理；
4. 包含异步代码路径的测试与调试指引。

当需要详细示例时，技能明确要求打开resources/implementation-playbook.md（仓库路径：implementation-playbook.md），下文所有模式均出自该文件。

二、核心概念：事件循环、协程、Task 与 Future

在进入代码模式之前，先建立对 asyncio 四大基础抽象的正确理解（对应 playbook 的 Core Concepts 章节）：

概念	说明	关键特征
事件循环（Event Loop）	asyncio 的心脏，负责管理与调度异步任务	单线程协作式多任务；调度协程执行；无阻塞地处理 I/O；管理回调和 Future
协程（Coroutine）	用async def定义的、可暂停可恢复的函数	通过await挂起/恢复；不会自行执行，必须被调度
Task	被调度到事件循环上并发运行的协程	通过asyncio.create_task()创建；独立调度执行
Future	表示异步操作最终结果的底层对象	是 Task 的底层机制；一般不需要直接操作

协程的基础语法：

async def my_coroutine(): result = await some_async_operation() return result

2.1 异步上下文管理器与异步迭代器

• Async Context Managers：支持async with语法，用于资源的正确清理（连接、会话、锁等），对应后面的 Pattern 6；
• Async 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–5）

Pattern 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": f"User {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(f"Fetched {len(users)} users") asyncio.run(main())

仓库真实佐证：skills/junta-leiloeiros/scripts/run_all.py（巴西商业登记处抓取编排器）正是这一模式的落地实现——它为 27 个州的抓取任务构建 task 列表后await asyncio.gather(*tasks)并发执行（run_all.py）：

semaphore = asyncio.Semaphore(concurrency) tasks = [scrape_state(uf, semaphore) for uf in estados_alvo] results = await asyncio.gather(*tasks)

Pattern 3：Task 的创建与管理

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 f"Result 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(f"Results: {result1}, {result2}") asyncio.run(main())

要点：create_task 返回的 Task 对象本身可await；如果创建后不等待也不取消，会在程序退出时收到 "Task was destroyed but it is pending" 告警，因此必须管理好每个 Task 的生命周期。

Pattern 4：异步代码的错误处理

asyncio.gather(..., return_exceptions=True)让异常以返回值形式收集到结果列表中，从而在聚合层面统一分流成功与失败：

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(f"Item {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(f"Error: {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_exceptions=True) # 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(f"Success: {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.TimeoutError：

import asyncio async def slow_operation(delay: int) -> str: """Operation that takes time.""" await asyncio.sleep(delay) return f"Completed after {delay}s" async def with_timeout(): """Execute operation with timeout.""" try: result = await asyncio.wait_for(slow_operation(5), timeout=2.0) print(result) except asyncio.TimeoutError: print("Operation timed out") asyncio.run(with_timeout())

四、进阶模式：资源控制与同步（Pattern 6–10）

Pattern 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(f"Using connection: {conn}") await asyncio.sleep(0.2) # Simulate query return {"rows": 10} asyncio.run(query_database())

Pattern 7：异步迭代器与生成器

用async for消费async def+yield定义的异步生成器，适合分页拉取等按需生产场景：

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": [f"item_{page}_{i}" for i in range(5)] } async def consume_async_iterator(): """Consume async iterator.""" async for number in async_range(1, 5): print(f"Number: {number}") print("\nFetching pages:") async for page_data in fetch_pages("https://api.example.com/items", 3): print(f"Page {page_data['page']}: {len(page_data['data'])} items") asyncio.run(consume_async_iterator())

Pattern 8：生产者-消费者模式（异步队列）

asyncio.Queue天然支持生产/消费解耦：生产者put数据，消费者get处理，None作为终止信号，queue.join()等待队列清空，最后显式取消消费者 Task：

import 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 = f"Item-{producer_id}-{i}" await queue.put(item) print(f"Producer {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(f"Consumer {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(maxsize=10) # 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(maxsize=10)本身就提供了背压（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(f"Calling {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 = [f"https://api.example.com/item/{i}" for i in range(20)] results = await rate_limited_requests(urls, max_concurrent=3) print(f"Completed {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(f"Worker {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(f"Final 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, timeout=aiohttp.ClientTimeout(total=10)) 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(total=10)在请求层实现了 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(f"SELECT * FROM users WHERE id = {user_id}") orders_task = db.execute(f"SELECT * FROM orders WHERE user_id = {user_id}") profile_task = db.fetch_one(f"SELECT * 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(f"Sending to {self.client_id}: {message}") await asyncio.sleep(0.01) async def recv(self) -> str: """Receive message.""" await asyncio.sleep(1) return f"Message 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(f"Client {websocket.client_id} connected") async def unregister(self, websocket: WebSocket): """Unregister client.""" self.clients.remove(websocket) print(f"Client {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 7（async 迭代器接收消息）、Pattern 2（gather 广播）与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(limit=100, limit_per_host=10) async with aiohttp.ClientSession(connector=connector) as session: tasks = [session.get(f"https://api.example.com/item/{i}") for i in range(50)] responses = await asyncio.gather(*tasks) return responses

6.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(f"Processed batch {i // batch_size + 1}") async def process_item(item: str): """Process single item.""" await asyncio.sleep(0.1) return f"Processed: {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 Pitfalls）

7.1 忘记 await

调用 async 函数不await，只会拿到一个未执行的协程对象：

# Wrong - returns coroutine object, doesn't 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-blocking

7.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 cancellation

7.4 混用同步与异步代码

async 函数内不能直接await（语法错误）；同步入口应通过asyncio.run()桥接：

# Wrong - can't 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), timeout=1.0)

九、最佳实践速查表

playbook 在最后给出了 10 条可直接落地的经验总结（implementation-playbook.md）：

1. 用asyncio.run()作为入口（Python 3.7+）；
2. 总是await协程使其真正执行；
3. 多任务并发优先用gather()；
4. 用 try/except 实现结构化错误处理；
5. 用超时防止操作无限挂起；
6. 连接池化以提升性能；
7. 异步代码中避免阻塞操作；
8. 用 Semaphore 做限流；
9. 正确处任务取消（Cancellation）；
10. 用 pytest-asyncio 测试异步代码。

十、仓库中的更多异步实战参考

除本技能外，agentic-awesome-skills 仓库中还有多个以 asyncio 为核心的真实实现可供对照学习：

• run_all.py：Semaphore 限流 + gather 并发的多目标抓取编排器，--concurrency参数直接控制并行度；
• voice-ai-engine-development：语音 Agent 引擎示例（如 complete_voice_engine.py），体现异步在实时流式场景的应用；
• instagram：脚本体系大量使用 asyncio 编排 API 调用（如 publish.py）；
• skill-sentinel 与 mcp-builder：分析/评估类工具中的异步调度。

结语

async-python-patterns技能的价值在于：它不是零散代码片段，而是一套"先澄清负载特征 → 再选择并发原语 → 后补超时/背压/错误处理 → 最后测试验证"的完整方法论。事件循环与协程提供了单线程内的高并发骨架，gather/Queue/Semaphore/Lock 分别解决聚合、生产消费、限流与同步问题，而超时、取消与连接池则是让系统在生产环境保持健壮的关键护栏。无论是构建 FastAPI 异步 API、aiohttp 爬虫还是 WebSocket 实时服务，都可以从这套模式库中直接取材组合；仓库中run_all.py等真实实现则证明了这套模式在规模化 I/O 编排场景中的落地价值。

赞分享

• AI 技能
• AI 插件

【免费下载链接】agentic-awesome-skills

AAS 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 skill

  AI 技能AI 插件

• redis-py异步编程指南：利用asyncio实现高性能非阻塞操作

  redis py异步编程指南：利用asyncio实现高性能非阻塞操作 你是否在处理高并发Redis操作时遇到过性能瓶颈？传统同步客户端会阻塞主线程，导致应用响应

  后端数据库客户端缓存

• 3 分钟快速上手 bypass-paywalls-clean-filters：新手安装与订阅完整教程

  3 分钟快速上手 bypass paywalls clean filters：新手安装与订阅完整教程 你是否遇到过这样的情况：好不容易打开一篇想看的新闻，却被半

  开发工具