Python 异步编程 asyncio

import asyncio

async def hello():
    """定义协程函数"""
    print("Hello")
    await asyncio.sleep(1)  # 模拟异步操作
    print("World")

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

import asyncio

async def fetch_data():
    """模拟获取数据"""
    print("开始获取数据...")
    await asyncio.sleep(2)  # 模拟网络请求
    print("数据获取完成")
    return {"data": "some data"}

async def process_data():
    """处理数据"""
    data = await fetch_data()  # 等待获取数据
    print(f"处理: {data}")
    return data

# 运行
result = asyncio.run(process_data())
print(f"结果: {result}")

import asyncio

async def task(name, delay):
    """异步任务"""
    print(f"任务 {name} 开始")
    await asyncio.sleep(delay)
    print(f"任务 {name} 完成")
    return f"{name} 的结果"

async def main():
    # 创建任务
    task1 = asyncio.create_task(task("A", 2))
    task2 = asyncio.create_task(task("B", 1))
    
    # 等待任务完成
    result1 = await task1
    result2 = await task2
    
    print(f"结果1: {result1}")
    print(f"结果2: {result2}")

asyncio.run(main())

import asyncio

async def task(name, delay):
    await asyncio.sleep(delay)
    return f"{name} 完成"

async def main():
    # 同时运行多个协程
    results = await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3),
    )
    print(f"所有结果: {results}")

asyncio.run(main())

import asyncio

async def task(name, delay):
    await asyncio.sleep(delay)
    return f"{name}"

async def main():
    tasks = [
        asyncio.create_task(task("A", 2)),
        asyncio.create_task(task("B", 1)),
        asyncio.create_task(task("C", 3)),
    ]
    
    # 等待所有完成
    done, pending = await asyncio.wait(tasks)
    
    for task_obj in done:
        print(f"完成: {task_obj.result()}")

asyncio.run(main())

import asyncio

async def fetch_url(url):
    """模拟获取 URL"""
    print(f"获取: {url}")
    await asyncio.sleep(1)  # 模拟网络延迟
    return f"内容: {url}"

async def main():
    urls = [
        "https://example.com/1",
        "https://example.com/2",
        "https://example.com/3",
    ]
    
    # 串行（慢）
    # for url in urls:
    #     result = await fetch_url(url)
    #     print(result)
    
    # 并发（快）
    tasks = [fetch_url(url) for url in urls]
    results = await asyncio.gather(*tasks)
    
    for result in results:
        print(result)

asyncio.run(main())

import asyncio

async def slow_task():
    await asyncio.sleep(5)
    return "完成"

async def main():
    try:
        # 设置 2 秒超时
        result = await asyncio.wait_for(slow_task(), timeout=2)
        print(result)
    except asyncio.TimeoutError:
        print("任务超时！")

asyncio.run(main())

import asyncio

async def long_task():
    try:
        while True:
            print("工作中...")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("任务被取消")
        raise  # 重新抛出

async def main():
    task = asyncio.create_task(long_task())
    
    await asyncio.sleep(3)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("确认任务已取消")

asyncio.run(main())

import asyncio

class AsyncCounter:
    """异步计数器"""
    
    def __init__(self, limit):
        self.limit = limit
        self.current = 0
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.current >= self.limit:
            raise StopAsyncIteration
        await asyncio.sleep(0.5)
        self.current += 1
        return self.current

async def main():
    async for num in AsyncCounter(5):
        print(f"计数: {num}")

asyncio.run(main())

import asyncio

class AsyncResource:
    """异步资源"""
    
    async def __aenter__(self):
        print("获取资源")
        await asyncio.sleep(1)
        return self
    
    async def __aexit__(self, exc_type, exc, tb):
        print("释放资源")
        await asyncio.sleep(0.5)
    
    async def do_something(self):
        print("使用资源")
        await asyncio.sleep(1)

async def main():
    async with AsyncResource() as resource:
        await resource.do_something()

asyncio.run(main())

import asyncio
import time

async def fetch_page(page_id):
    """模拟获取页面"""
    print(f"开始获取页面 {page_id}")
    await asyncio.sleep(1)  # 模拟网络请求
    print(f"页面 {page_id} 获取完成")
    return f"页面 {page_id} 的内容"

async def main():
    start = time.time()
    
    # 创建 10 个并发任务
    pages = range(10)
    tasks = [fetch_page(page) for page in pages]
    results = await asyncio.gather(*tasks)
    
    elapsed = time.time() - start
    print(f"\n获取 {len(results)} 个页面，耗时: {elapsed:.2f} 秒")
    print(f"平均每个页面: {elapsed/len(results):.2f} 秒")

asyncio.run(main())

import asyncio
import time

# 同步版本
def sync_task(n):
    time.sleep(1)
    return n * n

def sync_main():
    start = time.time()
    results = [sync_task(i) for i in range(5)]
    print(f"同步: {time.time() - start:.2f}s, 结果: {results}")

# 异步版本
async def async_task(n):
    await asyncio.sleep(1)
    return n * n

async def async_main():
    start = time.time()
    tasks = [async_task(i) for i in range(5)]
    results = await asyncio.gather(*tasks)
    print(f"异步: {time.time() - start:.2f}s, 结果: {results}")

# 运行对比
print("=== 同步 ===")
sync_main()

print("\n=== 异步 ===")
asyncio.run(async_main())