Python脚本如何规避协程任务阻塞

wen python案例 2026-06-27 235

本文目录导读：

Python脚本如何规避协程任务阻塞

使用异步版本替代同步操作
使用协程调度器（如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())