#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  # gather 会收集所有协程的返回值

async def main():
    # 批量并行，注意要用 * 解包列表（如果任务列表是动态生成的）
    results = await asyncio.gather(
        greet("Alice"),
        greet("Bob"),
        greet("Charlie")
    )
    print("📋 All greetings done, participants:", results)

asyncio.run(main())

import asyncio

async def slow_task(name: str, delay: int):
    print(f"⏳ Task {name} started, takes {delay}s")
    await asyncio.sleep(delay)
    print(f"✅ Task {name} completed")
    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"🎉 All tasks finished: {results}")

asyncio.run(main())

import asyncio

async def fetch_raw_page(host: str):
    """异步获取指定域名的首页原始数据"""
    print(f"🔍 Starting to fetch {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"✅ Finished fetching {host}, got {len(raw_response)} bytes")
    return raw_response

async def main():
    # 测试域名
    test_hosts = ["www.example.com", "www.github.com", "www.python.org"]
    # 批量异步获取
    raw_pages = await asyncio.gather(*[fetch_raw_page(host) for host in test_hosts])
    print(f"📦 Total fetched {len(raw_pages)} pages")

asyncio.run(main())

import asyncio

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

async def main():
    try:
        # 等待最多5秒
        result = await asyncio.wait_for(maybe_stuck_task(), timeout=5)
        print(result)
    except asyncio.TimeoutError:
        print("⏰ Task timed out!")

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"⏲️ Countdown: {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"⏲️ Countdown: {num}")

asyncio.run(main())