asyncio.Semaphore引发死锁问题，请求排查代码缺陷

核心问题点 #

1. 循环遍历tasks阻塞事件循环，导致工作task无法调度 #

你在每次创建新task后，都会遍历整个tasks列表检查是否有已完成的异常task。当jobs数量很大时（比如几百上千个），这个遍历操作会占用事件循环的大量时间，导致真正执行工作的task无法被调度运行，自然也就无法触发finally块里的信号量释放操作。后续的sem.acquire()会一直等待信号量，最终导致程序冻结或超时。

2. 信号量的acquire与release分离，存在时序风险 #

你在main协程中先调用sem.acquire()，再创建task，然后在task的finally块中执行sem.release()。这种分离的方式会导致信号量的控制权分散：如果task因为事件循环被阻塞而无法及时执行到finally块，信号量会一直被占用，后续的acquire操作就会无限等待。

3. 异常处理时未清理task，可能导致信号量泄漏 #

当你在循环中发现某个task抛出异常就立即raise，main协程会直接退出，但此时已创建的task可能还在运行或未被调度。虽然finally块理论上会执行release，但如果事件循环因为main退出而停止，部分task可能无法完成清理，导致信号量计数异常。

修复后的代码示例 #

async def main():
    sem = asyncio.Semaphore(8)
    tasks = []

    async def task(**params):
        # 将信号量的acquire和release放在同一个task上下文里
        async with sem:
            return await do_the_work(**params)

    # 先创建所有task（由信号量控制并发数）
    for params in jobs:
        tasks.append(asyncio.create_task(task(**params)))

    # 使用as_completed处理已完成的task，及时捕获异常，同时不阻塞事件循环
    for done_task in asyncio.as_completed(tasks):
        try:
            result = await done_task
            handle_result(result)
        except Exception as e:
            # 处理异常，可选：取消所有未完成的task
            for t in tasks:
                if not t.done():
                    t.cancel()
            raise e

    # 或者如果不需要逐个处理结果，直接用gather并设置return_exceptions=False
    # results = await asyncio.gather(*tasks, return_exceptions=False)
    # for result in results:
    #     handle_result(result)

修复说明 #

1. 将信号量操作移入task内部：使用async with sem自动管理信号量的acquire和release，确保每个task的信号量操作是原子性的，避免控制权分离的问题。
2. 使用asyncio.as_completed处理已完成的task：这种方式会在有task完成时立即处理，无需遍历整个列表，不会阻塞事件循环，同时能及时捕获异常。
3. 异常处理时清理未完成的task：当捕获到异常时，取消所有未完成的task，避免资源泄漏，同时确保程序能快速退出。

内容的提问来源于stack exchange，提问作者Achim