本文目录导读：
- 使用异步版本替代同步操作
- 使用协程调度器（如asyncio.create_task）
- 使用run_in_executor处理阻塞操作
- 使用信号量限制并发数量
- 设置超时机制
- 使用上下文管理器管理资源
- 监控和调试阻塞
- 完整示例：Web爬虫避免阻塞
在Python中规避协程任务阻塞,核心思路是避免在协程中执行同步阻塞操作，以下是几种有效策略：
使用异步版本替代同步操作
import asyncio
import time
# ❌ 错误：同步阻塞
async def bad_example():
time.sleep(1) # 阻塞整个事件循环
return "done"
# ✅ 正确：异步等待
async def good_example():
await asyncio.sleep(1) # 让出控制权
return "done"
常见同步操作替换：
time.sleep()→asyncio.sleep()requests.get()→aiohttp.ClientSession().get()open()→aiofiles.open()threading.Lock→asyncio.Lock
使用协程调度器（如asyncio.create_task）
import asyncio
async def task1():
print("Task 1 开始")
await asyncio.sleep(2)
print("Task 1 结束")
async def task2():
print("Task 2 开始")
await asyncio.sleep(1)
print("Task 2 结束")
async def main():
# 并发执行多个任务
t1 = asyncio.create_task(task1())
t2 = asyncio.create_task(task2())
# 等待所有任务完成
await asyncio.gather(t1, t2)
asyncio.run(main())
使用run_in_executor处理阻塞操作
将阻塞操作放到线程池或进程池中执行：
import asyncio
import time
async def blocking_op():
loop = asyncio.get_event_loop()
# 在线程池中执行阻塞操作
result = await loop.run_in_executor(
None, # 使用默认线程池
time.sleep, # 阻塞函数
2 # 参数
)
return result
# 自定义线程池
async def with_custom_executor():
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=4)
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
executor,
time.sleep,
2
)
return result
使用信号量限制并发数量
防止过载导致阻塞：
import asyncio
async def limited_requests():
semaphore = asyncio.Semaphore(10) # 最多10个并发
async def fetch_url(url):
async with semaphore:
# 异步请求
pass
tasks = [fetch_url(f"url_{i}") for i in range(100)]
await asyncio.gather(*tasks)
设置超时机制
防止任务长期阻塞：
import asyncio
async def task_with_timeout():
try:
result = await asyncio.wait_for(
async_operation(),
timeout=5.0 # 5秒超时
)
return result
except asyncio.TimeoutError:
print("任务超时")
return None
async def async_operation():
await asyncio.sleep(10)
return "完成"
使用上下文管理器管理资源
import aiohttp
import asyncio
async def safe_http_request():
# 使用with确保资源正确释放
async with aiohttp.ClientSession() as session:
async with session.get('http://example.com') as response:
return await response.text()
监控和调试阻塞
import asyncio
import signal
async def monitor_blocking():
loop = asyncio.get_event_loop()
# 检测长阻塞操作
loop.slow_callback_duration = 0.1 # 超过0.1秒记录警告
# 使用调试模式
loop.set_debug(True)
# 检查事件循环状态
async def check_event_loop():
while True:
loop = asyncio.get_event_loop()
print(f"待处理任务: {len(asyncio.all_tasks(loop))}")
await asyncio.sleep(1)
完整示例：Web爬虫避免阻塞
import asyncio
import aiohttp
from typing import List
class AsyncWebCrawler:
def __init__(self, max_concurrent=10):
self.semaphore = asyncio.Semaphore(max_concurrent)
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, *args):
await self.session.close()
async def fetch(self, url: str, timeout: float = 10.0):
async with self.semaphore:
try:
async with self.session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as resp:
return await resp.text()
except Exception as e:
print(f"请求失败: {url}, 错误: {e}")
return None
async def crawl_many(self, urls: List[str]):
tasks = [self.fetch(url) for url in urls]
return await asyncio.gather(*tasks)
async def main():
urls = [f"http://example.com/page/{i}" for i in range(20)]
async with AsyncWebCrawler(max_concurrent=5) as crawler:
results = await crawler.crawl_many(urls)
# 处理结果...
asyncio.run(main())
- 避免同步调用：始终使用异步版本的操作
- 合理控制并发：使用信号量或连接池限制并发数
- 设置超时：防止单个任务长时间阻塞
- 分离阻塞操作：使用
run_in_executor处理无法避免的同步操作 - 监控性能：启用调试模式检测异常阻塞
- 正确异常处理：捕获和处理超时、连接错误等异常
遵循这些原则可以显著提高协程程序的并发性能和稳定性。
