#Python asyncio 异步编程教程

async def my_first_coro():
    print("我是协程")

import asyncio

async def hello():
    print("Hello world!")
    # 模拟一个耗时 1 秒的 I/O 操作（例如读文件、网络请求）
    await asyncio.sleep(1)
    print("Hello again!")

# 启动顶层协程
asyncio.run(hello())

import asyncio

async def greet(name: str):
    print(f"👋 Hello {name}!")
    await asyncio.sleep(1)   # 模拟给不同人发消息的延迟
    print(f"👋 Goodbye {name}!")
    return name

async def main():
    # gather 会收集所有协程的返回值
    results = await asyncio.gather(
        greet("Alice"),
        greet("Bob"),
        greet("Charlie")
    )
    print("📋 所有问候完成，参与者：", results)

asyncio.run(main())

import asyncio

async def slow_task(name: str, delay: int):
    print(f"⏳ 任务 {name} 启动，预计 {delay}s 完成")
    await asyncio.sleep(delay)
    print(f"✅ 任务 {name} 完成")
    return name

async def main():
    # 立即创建并启动两个任务
    task_a = asyncio.create_task(slow_task("A", 2))
    task_b = asyncio.create_task(slow_task("B", 1))

    # 等待所有任务完成
    results = await asyncio.gather(task_a, task_b)
    print(f"🎉 所有任务完成：{results}")

asyncio.run(main())

import asyncio

async def fetch_raw_page(host: str):
    """异步获取指定域名的首页原始数据"""
    print(f"🔍 开始获取 {host} ...")
    # 建立异步 TCP 连接
    reader, writer = await asyncio.open_connection(host, 80)

    # 构造 HTTP GET 请求
    request = f"GET / HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n"
    writer.write(request.encode("utf-8"))
    # 确保请求完全发送
    await writer.drain()

    # 异步读取响应内容
    raw_response = await reader.read()
    # 关闭连接
    writer.close()
    await writer.wait_closed()

    print(f"✅ 获取 {host} 完成，接收 {len(raw_response)} 字节")
    return raw_response

async def main():
    hosts = ["www.example.com", "www.github.com", "www.python.org"]
    # 批量异步获取多个页面
    raw_pages = await asyncio.gather(*[fetch_raw_page(host) for host in hosts])
    print(f"📦 总共获取了 {len(raw_pages)} 个页面")

asyncio.run(main())

import asyncio

async def maybe_stuck_task():
    await asyncio.sleep(10)   # 假装一个会卡死的任务
    return "Success"

async def main():
    try:
        result = await asyncio.wait_for(maybe_stuck_task(), timeout=5)
        print(result)
    except asyncio.TimeoutError:
        print("⏰ 任务超时！")

asyncio.run(main())

import asyncio

class AsyncTimer:
    """异步倒计时器"""
    def __init__(self, start: int):
        self.current = start

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.current <= 0:
            raise StopAsyncIteration
        await asyncio.sleep(1)
        self.current -= 1
        return self.current + 1

async def main():
    async for num in AsyncTimer(3):
        print(f"⏲️ 倒计时：{num}")

asyncio.run(main())

import asyncio

async def async_timer_gen(start: int):
    """异步倒计时生成器（更简洁）"""
    for num in range(start, 0, -1):
        await asyncio.sleep(1)
        yield num

async def main():
    async for num in async_timer_gen(3):
        print(f"⏲️ 倒计时：{num}")

asyncio.run(main())