Python async案例如何定义异步函数

wen python案例 2026-06-30 214

本文目录导读：

基本异步函数定义
多个异步函数并发执行
带参数的异步函数
异步上下文管理器
超时控制的异步函数
完整的实际案例：异步Web请求

我来为你介绍Python中定义异步函数的方法和完整案例。

基本异步函数定义

import asyncio
# 使用 async def 定义异步函数
async def hello():
    print("Hello")
    await asyncio.sleep(1)  # 模拟异步操作
    print("World")
    return "完成"
# 运行异步函数
async def main():
    result = await hello()
    print(result)
# 启动事件循环
asyncio.run(main())

多个异步函数并发执行

import asyncio
import time
async def task(name, seconds):
    print(f"任务 {name} 开始")
    await asyncio.sleep(seconds)  # 模拟耗时操作
    print(f"任务 {name} 完成，耗时 {seconds}秒")
    return f"任务{name}的结果"
async def main():
    start = time.time()
    # 并发执行多个任务
    tasks = [
        task("A", 2),
        task("B", 3),
        task("C", 1)
    ]
    # 等待所有任务完成
    results = await asyncio.gather(*tasks)
    print(f"总耗时: {time.time() - start:.2f}秒")
    print(f"结果: {results}")
asyncio.run(main())

带参数的异步函数

import asyncio
async def fetch_data(url, timeout=10):
    """
    模拟异步获取数据
    """
    print(f"正在从 {url} 获取数据...")
    await asyncio.sleep(timeout)  # 模拟网络请求
    # 模拟返回的数据
    data = {
        "url": url,
        "status": 200,
        "content": f"这是来自 {url} 的数据"
    }
    return data
async def process_data(data):
    """处理数据的异步函数"""
    print(f"处理数据: {data['url']}")
    await asyncio.sleep(0.5)
    data["processed"] = True
    return data
async def main():
    # 获取数据
    raw_data = await fetch_data("https://api.example.com/users", 2)
    # 处理数据
    processed_data = await process_data(raw_data)
    print(f"最终结果: {processed_data}")
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_val, exc_tb):
        print("异步关闭资源")
        await asyncio.sleep(0.5)
    async def do_work(self):
        print("执行异步工作")
        await asyncio.sleep(0.5)
        return "工作完成"
async def main():
    async with AsyncResource() as resource:
        result = await resource.do_work()
        print(result)
asyncio.run(main())

超时控制的异步函数

import asyncio
async def long_running_task():
    print("开始长时间运行的任务")
    await asyncio.sleep(10)  # 模拟长时间操作
    return "任务结果"
async def main():
    try:
        # 设置超时时间
        result = await asyncio.wait_for(long_running_task(), timeout=3)
        print(result)
    except asyncio.TimeoutError:
        print("任务超时！")
asyncio.run(main())

完整的实际案例：异步Web请求

import asyncio
import aiohttp  # 需要安装: pip install aiohttp
async def fetch_url(session, url):
    """异步获取URL内容"""
    try:
        async with session.get(url) as response:
            if response.status == 200:
                content = await response.text()
                return {"url": url, "status": response.status, "length": len(content)}
            else:
                return {"url": url, "status": response.status, "error": "请求失败"}
    except Exception as e:
        return {"url": url, "status": 0, "error": str(e)}
async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/3",
    ]
    async with aiohttp.ClientSession() as session:
        # 并发请求多个URL
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(f"URL: {result['url']}")
            print(f"状态码: {result.get('status')}")
            print(f"内容长度: {result.get('length', 'N/A')}")
            print("-" * 50)
asyncio.run(main())