Python异步编程之await与asyncio基本用法详解

async def my_coroutine():
    await asyncio.sleep(1)
    print("Done")

async def fetch_data(url):
    # 模拟网络请求
    await asyncio.sleep(1)
    return f"Data from {url}"

async def main():
    result = await fetch_data("https://example.com")
    print(result)
# Python 3.7+ 推荐方式
asyncio.run(main())

async def main():
    # 同时执行多个协程
    task1 = asyncio.create_task(fetch_data("url1"))
    task2 = asyncio.create_task(fetch_data("url2"))
    # 等待所有任务完成
    results = await asyncio.gather(task1, task2)
    print(results)

async def limited_fetch(url, semaphore):
    async with semaphore:
        return await fetch_data(url)
async def main():
    semaphore = asyncio.Semaphore(5)  # 最大并发5
    tasks = [limited_fetch(url, semaphore) for url in urls]
    await asyncio.gather(*tasks)

async def fetch_with_timeout():
    try:
        async with asyncio.timeout(3):  # Python 3.11+
            await fetch_data("slow_url")
    except TimeoutError:
        print("Timeout!")

async def main():
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    def callback():
        future.set_result("Done")
    loop.call_soon(callback)
    result = await future
    print(result)

async def main():
    fetch_data("url")  # 错误！没有 await

async def bad_example():
    time.sleep(1)  # 错误！应使用 await asyncio.sleep(1)

import asyncio
async def download(url, delay):
    print(f"Start downloading {url}")
    await asyncio.sleep(delay)
    print(f"Finished {url}")
    return url
async def main():
    urls = [
        ("https://site1.com", 1),
        ("https://site2.com", 2),
        ("https://site3.com", 3),
    ]
    tasks = [asyncio.create_task(download(url, delay)) for url, delay in urls]
    results = await asyncio.gather(*tasks)
    print("All done:", results)
if __name__ == "__main__":
    asyncio.run(main())