Python asyncio 异步编程

# 传统的同步方式
import time

def download_data(url):
    print(f"开始下载 {url}")
    time.sleep(2)  # 模拟网络延迟
    print(f"下载完成 {url}")
    return f"{url} 的数据"

# 顺序执行，总共需要6秒
start = time.time()
result1 = download_data("https://fly63.com/page1")
result2 = download_data("https://fly63.com/page2") 
result3 = download_data("https://fly63.com/page3")
end = time.time()
print(f"总耗时: {end - start} 秒")

# 异步方式
import asyncio

async def download_data(url):
    print(f"开始下载 {url}")
    await asyncio.sleep(2)  # 模拟网络延迟，但不会阻塞
    print(f"下载完成 {url}")
    return f"{url} 的数据"

async def main():
    start = time.time()
    # 同时启动三个下载任务
    task1 = asyncio.create_task(download_data("https://fly63.com/page1"))
    task2 = asyncio.create_task(download_data("https://fly63.com/page2"))
    task3 = asyncio.create_task(download_data("https://fly63.com/page3"))
    
    # 等待所有任务完成
    results = await asyncio.gather(task1, task2, task3)
    end = time.time()
    print(f"总耗时: {end - start} 秒")
    print(f"结果: {results}")

asyncio.run(main())

import asyncio

async def my_coroutine():
    print("开始执行")
    await asyncio.sleep(1)  # 暂停1秒
    print("1秒后继续执行")
    await asyncio.sleep(1)  # 再暂停1秒
    print("又过了1秒")
    return "完成"

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

import asyncio

async def task1():
    for i in range(3):
        print(f"任务1: 第{i+1}次执行")
        await asyncio.sleep(0.5)

async def task2():
    for i in range(2):
        print(f"任务2: 第{i+1}次执行") 
        await asyncio.sleep(0.8)

async def main():
    # 同时运行两个任务
    await asyncio.gather(task1(), task2())

asyncio.run(main())

import asyncio

async def count_down(name, seconds):
    for i in range(seconds, 0, -1):
        print(f"{name}: {i}秒")
        await asyncio.sleep(1)
    print(f"{name}: 完成!")
    return f"{name}的结果"

async def main():
    # 创建多个任务
    task1 = asyncio.create_task(count_down("任务A", 3))
    task2 = asyncio.create_task(count_down("任务B", 5))
    task3 = asyncio.create_task(count_down("任务C", 2))
    
    # 等待所有任务完成并获取结果
    results = await asyncio.gather(task1, task2, task3)
    print(f"所有任务完成: {results}")

asyncio.run(main())

import asyncio
import aiohttp  # 需要安装: pip install aiohttp

async def download_page(session, url):
    try:
        print(f"开始下载: {url}")
        async with session.get(url, timeout=10) as response:
            content = await response.text()
            print(f"下载完成: {url}, 长度: {len(content)}")
            return len(content)
    except Exception as e:
        print(f"下载失败 {url}: {e}")
        return 0

async def main():
    urls = [
        "https://www.fly63.com",
        "https://www.fly63.com/python",
        "https://www.fly63.com/javascript", 
        "https://www.fly63.com/css",
        "https://www.fly63.com/html"
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [download_page(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        
        total_size = sum(results)
        print(f"总共下载 {len(urls)} 个页面，总大小: {total_size} 字符")

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))
    
    # 发送结束信号
    await queue.put(None)

async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            # 把结束信号放回队列，让其他消费者也能看到
            await queue.put(None)
            print(f"消费者 {name} 结束工作")
            break
            
        print(f"消费者 {name} 处理了: {item}")
        await asyncio.sleep(random.uniform(0.2, 0.8))
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=5)
    
    # 创建生产者和消费者
    producers = [
        asyncio.create_task(producer(queue, "A", 3)),
        asyncio.create_task(producer(queue, "B", 2))
    ]
    
    consumers = [
        asyncio.create_task(consumer(queue, "X")),
        asyncio.create_task(consumer(queue, "Y"))
    ]
    
    # 等待所有生产者完成
    await asyncio.gather(*producers)
    
    # 等待队列中的所有项目被处理
    await queue.join()
    
    # 取消消费者任务
    for consumer_task in consumers:
        consumer_task.cancel()

asyncio.run(main())

import asyncio

async def slow_operation(seconds):
    print(f"开始执行需要{seconds}秒的操作")
    await asyncio.sleep(seconds)
    return f"操作完成，耗时{seconds}秒"

async def main():
    try:
        # 如果操作超过3秒就超时
        result = await asyncio.wait_for(slow_operation(5), timeout=3.0)
        print(result)
    except asyncio.TimeoutError:
        print("操作超时！")
    
    # 另一种方式：使用shield防止任务被取消
    try:
        result = await asyncio.wait_for(
            asyncio.shield(slow_operation(5)), 
            timeout=3.0
        )
        print(result)
    except asyncio.TimeoutError:
        print("超时了，但任务仍在后台运行")
        # 等待原始任务完成
        await asyncio.sleep(2)
        print("现在原始任务应该完成了")

asyncio.run(main())

import asyncio

class BankAccount:
    def __init__(self, balance):
        self.balance = balance
        self.lock = asyncio.Lock()
    
    async def withdraw(self, amount, user):
        async with self.lock:  # 确保同一时间只有一个协程可以修改余额
            print(f"{user} 尝试取款 {amount}")
            if self.balance >= amount:
                await asyncio.sleep(0.1)  # 模拟处理时间
                self.balance -= amount
                print(f"{user} 取款成功，余额: {self.balance}")
                return True
            else:
                print(f"{user} 取款失败，余额不足")
                return False

async def user_operation(account, user_name):
    for _ in range(3):
        success = await account.withdraw(10, user_name)
        await asyncio.sleep(0.2)

async def main():
    account = BankAccount(50)
    
    # 多个用户同时操作同一个账户
    await asyncio.gather(
        user_operation(account, "用户A"),
        user_operation(account, "用户B"),
        user_operation(account, "用户C")
    )
    
    print(f"最终余额: {account.balance}")

asyncio.run(main())

import asyncio
import time

def sync_blocking_operation():
    """这是一个同步的阻塞操作"""
    time.sleep(2)  # 这会阻塞整个事件循环！
    return "同步操作完成"

async def async_wrapper():
    """将同步操作包装成异步"""
    # 在线程池中运行同步函数，避免阻塞事件循环
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, sync_blocking_operation)
    return result

async def main():
    print("开始执行")
    result = await async_wrapper()
    print(f"结果: {result}")
    
    # 同时执行其他异步任务
    task1 = asyncio.create_task(async_wrapper())
    task2 = asyncio.create_task(asyncio.sleep(1))
    
    results = await asyncio.gather(task1, task2)
    print("所有任务完成")

asyncio.run(main())

import asyncio

async def might_fail_operation(name, success_chance):
    await asyncio.sleep(0.5)
    if success_chance < 0.5:
        raise ValueError(f"{name} 操作失败！")
    return f"{name} 操作成功"

async def main():
    tasks = [
        asyncio.create_task(might_fail_operation("任务A", 0.8)),
        asyncio.create_task(might_fail_operation("任务B", 0.3)),
        asyncio.create_task(might_fail_operation("任务C", 0.9))
    ]
    
    # 使用return_exceptions=True让gather不因单个异常而停止
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"任务{i}出错: {result}")
        else:
            print(f"任务{i}成功: {result}")

asyncio.run(main())

import asyncio

async def limited_operation(semaphore, name):
    async with semaphore:
        print(f"{name} 开始执行")
        await asyncio.sleep(1)
        print(f"{name} 完成")
        return f"{name} 的结果"

async def main():
    # 限制同时只能有2个操作运行
    semaphore = asyncio.Semaphore(2)
    
    tasks = []
    for i in range(6):
        task = asyncio.create_task(limited_operation(semaphore, f"任务{i}"))
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)
    print("所有任务完成")

asyncio.run(main())