如何在不阻塞线程的情况下运行多个周期性Python协程？

你的问题核心是无限循环的协程导致asyncio.gather永远阻塞，而如果不等待协程，ClientSession会提前关闭引发错误。要实现可取消且不阻塞的运行，需要给协程添加取消触发机制，同时确保session在所有协程结束后再关闭。

具体实现步骤 #

• 给协程添加取消检测：在循环中定期检查任务是否被取消，主动退出循环。
• 在main函数中添加取消触发逻辑：比如监听用户（按Ctrl+C）的终止信号，或者设置自动停止时长。
• 取消所有任务后，等待任务完成取消流程，再退出session上下文。

修改后的可用性检查代码 #

import asyncio
import aiohttp
import signal


LIMIT = 2

async def check_uptime_coro(session: aiohttp.ClientSession, url: str, semaphore: asyncio.BoundedSemaphore) -> None:
    try:
        while True:
            # 检查任务是否被取消，若已取消则主动退出
            if asyncio.current_task().cancelled():
                break
            
            try:
                async with semaphore:
                    async with session.get(url) as response:
                        if response.status != 200:
                            print(f"error with {url} {response.status}")
                        else:
                            print(f"success with {url}")
                # 休眠期间响应取消操作
                await asyncio.sleep(5)
            except Exception as e:
                print(f"error with {url} {e}")
    except asyncio.CancelledError:
        print(f"Stopping check for {url}")
        raise  # 重新抛出异常，让gather感知任务取消状态


async def main() -> None:
    urls = [f"http://localhost:8080/{x}" for x in range(0, 10)]
    tasks = {}
    semaphore = asyncio.BoundedSemaphore(LIMIT)
    
    # 捕获Ctrl+C信号触发停止
    loop = asyncio.get_running_loop()
    stop_event = asyncio.Event()
    
    def handle_signal():
        stop_event.set()
    
    loop.add_signal_handler(signal.SIGINT, handle_signal)
    
    try:
        async with aiohttp.ClientSession() as session:
            for url in urls:
                tasks[url] = asyncio.create_task(
                    check_uptime_coro(session, url, semaphore))
            
            # 等待停止信号（用户按Ctrl+C）
            await stop_event.wait()
            
            # 取消所有任务
            for task in tasks.values():
                task.cancel()
            
            # 等待所有任务完成取消流程
            await asyncio.gather(*tasks.values(), return_exceptions=True)
            
        print("All tasks stopped and session closed properly!")
    except Exception as e:
        print(f"error! {e}")


if __name__ == "__main__":
    asyncio.run(main())
    print("Script exited successfully!")

关键说明 #

• 取消响应：协程中通过asyncio.current_task().cancelled()主动检查取消状态，await asyncio.sleep()也会响应取消操作抛出异常，确保协程能及时停止。
• 优雅关闭：先触发所有任务取消，再通过asyncio.gather等待任务完成取消流程，避免session提前关闭导致的报错。
• 自定义停止逻辑：如果不需要手动触发，也可以把await stop_event.wait()替换为await asyncio.sleep(30)，实现固定时长后自动停止。

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