Python asyncio异步编程完全指南：事件循环与协程实战 - 小确幸生活

import asyncio

# 定义协程函数
async def my_coroutine():
    print("开始执行")
    await asyncio.sleep(1)  # 异步等待
    print("执行完成")
    return "结果"

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

import asyncio
import time

async def say_hello(delay, name):
    """异步函数"""
    print(f"[{name}] 开始，等待 {delay} 秒")
    await asyncio.sleep(delay)  # 非阻塞等待
    print(f"[{name}] 你好！")
    return f"{name} 完成"

# 运行单个协程
result = asyncio.run(say_hello(1, "Alice"))
print(f"结果: {result}")

import asyncio
import time

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

async def main():
    # 方式1: 使用 gather 并发执行
    start = time.time()
    results = await asyncio.gather(
        task("任务1", 2),
        task("任务2", 1),
        task("任务3", 3)
    )
    print(f"总耗时: {time.time() - start:.2f} 秒")  # 约 3 秒
    print(f"所有结果: {results}")

asyncio.run(main())

import asyncio

async def background_task(name, delay):
    print(f"[{name}] 开始")
    await asyncio.sleep(delay)
    print(f"[{name}] 完成")
    return f"{name} 的结果"

async def main():
    # 创建任务（立即开始执行）
    task1 = asyncio.create_task(background_task("任务A", 2))
    task2 = asyncio.create_task(background_task("任务B", 1))
    
    # 可以做其他事情
    print("主程序继续执行...")
    await asyncio.sleep(0.5)
    print("主程序等待任务完成...")
    
    # 等待任务完成
    result1 = await task1
    result2 = await task2
    
    print(f"结果1: {result1}")
    print(f"结果2: {result2}")

asyncio.run(main())

import asyncio
import random

async def fetch_data(url, delay):
    """模拟异步网络请求"""
    print(f"开始请求: {url}")
    await asyncio.sleep(delay)  # 模拟网络延迟
    data = {"url": url, "status": 200, "data": f"数据_{random.randint(1, 100)}"}
    print(f"完成请求: {url}")
    return data

async def main():
    urls = [
        ("https://api.example.com/1", 2),
        ("https://api.example.com/2", 1),
        ("https://api.example.com/3", 3),
        ("https://api.example.com/4", 1.5),
    ]
    
    # 并发请求所有 URL
    tasks = [fetch_data(url, delay) for url, delay in urls]
    results = await asyncio.gather(*tasks)
    
    print("\n所有结果:")
    for result in results:
        print(f"  {result}")

asyncio.run(main())

import asyncio
import random

async def producer(queue, name, count):
    """生产者：向队列添加数据"""
    for i in range(count):
        item = f"{name}-产品-{i}"
        await queue.put(item)
        print(f"[生产者 {name}] 生产: {item}")
        await asyncio.sleep(random.uniform(0.1, 0.5))
    print(f"[生产者 {name}] 完成生产")

async def consumer(queue, name):
    """消费者：从队列获取数据"""
    while True:
        item = await queue.get()
        if item is None:  # 结束信号
            break
        print(f"[消费者 {name}] 消费: {item}")
        await asyncio.sleep(random.uniform(0.2, 0.6))
        queue.task_done()
    print(f"[消费者 {name}] 完成消费")

async def main():
    # 创建队列
    queue = asyncio.Queue(maxsize=10)
    
    # 创建生产者和消费者
    producers = [
        producer(queue, "P1", 5),
        producer(queue, "P2", 5),
    ]
    consumers = [
        consumer(queue, "C1"),
        consumer(queue, "C2"),
    ]
    
    # 并发执行
    await asyncio.gather(*producers)
    
    # 发送结束信号
    for _ in consumers:
        await queue.put(None)
    
    await asyncio.gather(*consumers)
    print("所有任务完成")

asyncio.run(main())

import asyncio

async def slow_operation():
    """慢速操作"""
    await asyncio.sleep(5)
    return "完成"

async def main():
    try:
        # 设置超时时间为 2 秒
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
        print(f"操作成功: {result}")
    except asyncio.TimeoutError:
        print("操作超时！")

asyncio.run(main())