Python：async_timeout 库高级用法举例和应用详解

import asyncio  # 导入asyncio库进行异步操作
import aiohttp  # 导入aiohttp库用于发起HTTP请求
import async_timeout  # 导入async_timeout库以设置超时

async def fetch_data(url):
    async with aiohttp.ClientSession() as session:  # 创建Session对象
        try:
            # 使用async_timeout设置5秒的超时
            async with async_timeout.timeout(5):  
                async with session.get(url) as response:  # 发起GET请求
                    return await response.json()  # 返回JSON格式的数据
        except asyncio.TimeoutError:  # 捕获超时异常
            print("请求超时!")  # 输出超时信息

# 运行fetch_data函数
asyncio.run(fetch_data("https://jsonplaceholder.typicode.com/todos/1"))

import asyncio  # 导入asyncio库进行异步操作
import async_timeout  # 导入async_timeout库以设置超时
import aiomysql  # 导入aiomysql库用于异步MySQL操作

async def fetch_from_db(query):
    # 创建数据库连接
    conn = await aiomysql.connect(user='user', password='password', db='test_db',
                                   host='127.0.0.1', port=3306)
    try:
        async with conn.cursor() as cursor:  # 创建游标
            try:
                # 使用async_timeout设置10秒的超时
                async with async_timeout.timeout(10):
                    await cursor.execute(query)  # 执行查询
                    result = await cursor.fetchall()  # 获取查询结果
                    return result
            except asyncio.TimeoutError:  # 捕获超时异常
                print("数据库操作超时!")  # 输出超时信息
    finally:
        conn.close()  # 确保关闭数据库连接

# 运行fetch_from_db函数
asyncio.run(fetch_from_db("SELECT * FROM users"))

import asyncio  # 导入asyncio库进行异步操作
import async_timeout  # 导入async_timeout库以设置超时

async def task(n):
    # 模拟异步任务
    await asyncio.sleep(n)  # 睡眠n秒表示耗时操作
    return f"任务{n}完成"  # 返回任务完成信息

async def main():
    tasks = [task(1), task(2), task(3)]  # 创建多个异步任务
    try:
        async with async_timeout.timeout(5):  # 设置总超时5秒
            results = await asyncio.gather(*tasks)  # 并发执行任务
            print(results)  # 输出任务结果
    except asyncio.TimeoutError:  # 捕获超时异常
        print("某些任务超时!")  # 输出超时信息

# 运行main函数
asyncio.run(main())