协程 #

import asyncio

# 定义协程
async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

# 运行协程
asyncio.run(hello())

import asyncio

async def fetch_data(url):
    print(f"获取 {url}")
    await asyncio.sleep(1)  # 模拟IO操作
    return f"{url} 的数据"

async def main():
    result = await fetch_data("https://example.com")
    print(result)

asyncio.run(main())

import asyncio

async def task(name, seconds):
    print(f"{name} 开始")
    await asyncio.sleep(seconds)
    print(f"{name} 完成")
    return f"{name} 结果"

async def main():
    # 并发执行多个任务
    results = await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3)
    )
    print(results)

asyncio.run(main())

import asyncio

async def main():
    # 创建任务
    task1 = asyncio.create_task(some_coroutine())
    task2 = asyncio.create_task(another_coroutine())
    
    # 等待任务完成
    await task1
    await task2
    
    # 或使用gather
    await asyncio.gather(task1, task2)

import asyncio

async def long_operation():
    await asyncio.sleep(10)
    return "完成"

async def main():
    try:
        result = await asyncio.wait_for(
            long_operation(), 
            timeout=2.0
        )
    except asyncio.TimeoutError:
        print("操作超时")

import asyncio

async def async_generator():
    for i in range(5):
        await asyncio.sleep(0.1)
        yield i

async def main():
    # 异步迭代
    async for item in async_generator():
        print(item)
    
    # 异步推导式
    result = [item async for item in async_generator()]

import asyncio

class AsyncContext:
    async def __aenter__(self):
        print("进入")
        return self
    
    async def __aexit__(self, *args):
        print("退出")

async def main():
    async with AsyncContext() as ctx:
        await asyncio.sleep(1)

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://example.com",
        "https://python.org"
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(result[:100])

asyncio.run(main())