Python 异步编程高级应用指南

第一程序员

464人浏览 · 2026-04-27 21:59:16

第一程序员 · 2026-04-27 21:59:16 发布

Python 异步编程高级应用指南

1. 异步编程基础

异步编程是一种编程范式，它允许我们在等待某个操作完成时执行其他任务，从而提高程序的性能和响应速度。

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

# 运行异步函数
asyncio.run(hello())

2. 协程和任务

2.1 协程

协程是异步编程的基本单位，它是一种可以暂停执行并在稍后恢复的函数。

import asyncio

async def coroutine1():
    print("Coroutine 1 started")
    await asyncio.sleep(1)
    print("Coroutine 1 finished")

async def coroutine2():
    print("Coroutine 2 started")
    await asyncio.sleep(0.5)
    print("Coroutine 2 finished")

async def main():
    await coroutine1()
    await coroutine2()

# 运行主协程
asyncio.run(main())

2.2 任务

任务是协程的包装器，它允许我们并行执行多个协程。

import asyncio

async def coroutine1():
    print("Coroutine 1 started")
    await asyncio.sleep(1)
    print("Coroutine 1 finished")
    return "Result from coroutine 1"

async def coroutine2():
    print("Coroutine 2 started")
    await asyncio.sleep(0.5)
    print("Coroutine 2 finished")
    return "Result from coroutine 2"

async def main():
    # 创建任务
    task1 = asyncio.create_task(coroutine1())
    task2 = asyncio.create_task(coroutine2())
    
    # 等待任务完成
    result1 = await task1
    result2 = await task2
    
    print(f"Results: {result1}, {result2}")

# 运行主协程
asyncio.run(main())

3. 高级异步编程技巧

3.1 并行执行多个任务

import asyncio

async def fetch_data(url):
    print(f"Fetching data from {url}")
    await asyncio.sleep(1)
    return f"Data from {url}"

async def main():
    # 并行执行多个任务
    urls = ["https://example.com", "https://google.com", "https://github.com"]
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    
    for result in results:
        print(result)

# 运行主协程
asyncio.run(main())

3.2 超时处理

import asyncio

async def slow_operation():
    print("Starting slow operation")
    await asyncio.sleep(2)
    print("Slow operation finished")
    return "Result"

async def main():
    try:
        # 设置超时
        result = await asyncio.wait_for(slow_operation(), timeout=1)
        print(f"Result: {result}")
    except asyncio.TimeoutError:
        print("Operation timed out")

# 运行主协程
asyncio.run(main())

3.3 任务取消

import asyncio

async def long_running_task():
    try:
        print("Starting long running task")
        for i in range(10):
            print(f"Working... {i}")
            await asyncio.sleep(0.5)
        print("Task finished")
    except asyncio.CancelledError:
        print("Task was cancelled")
        raise

async def main():
    task = asyncio.create_task(long_running_task())
    
    # 等待一段时间后取消任务
    await asyncio.sleep(2)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("Main caught cancelled error")

# 运行主协程
asyncio.run(main())

4. 异步IO操作

4.1 异步文件操作

import asyncio
import aiofiles

async def read_file():
    async with aiofiles.open('file.txt', 'r') as f:
        content = await f.read()
        print(f"File content: {content}")

async def write_file():
    async with aiofiles.open('output.txt', 'w') as f:
        await f.write('Hello, World!')
        print("File written")

async def main():
    await read_file()
    await write_file()

# 运行主协程
asyncio.run(main())

4.2 异步网络操作

import asyncio
import aiohttp

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    url = "https://example.com"
    content = await fetch_url(url)
    print(f"URL content length: {len(content)}")

# 运行主协程
asyncio.run(main())

5. 异步上下文管理器

import asyncio

class AsyncContextManager:
    async def __aenter__(self):
        print("Entering context")
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Exiting context")

async def main():
    async with AsyncContextManager():
        print("Inside context")
        await asyncio.sleep(1)

# 运行主协程
asyncio.run(main())

6. 实际应用场景

6.1 异步Web服务器

from aiohttp import web

async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    await asyncio.sleep(1)  # 模拟异步操作
    return web.Response(text=f"Hello, {name}!")

app = web.Application()
app.add_routes([
    web.get('/', handle),
    web.get('/{name}', handle)
])

if __name__ == '__main__':
    web.run_app(app)

6.2 异步爬虫

import asyncio
import aiohttp
from bs4 import BeautifulSoup

async def fetch_page(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def parse_page(url):
    content = await fetch_page(url)
    soup = BeautifulSoup(content, 'html.parser')
    title = soup.title.string
    print(f"Title of {url}: {title}")

async def main():
    urls = [
        "https://example.com",
        "https://google.com",
        "https://github.com"
    ]
    tasks = [parse_page(url) for url in urls]
    await asyncio.gather(*tasks)

# 运行主协程
asyncio.run(main())

6.3 异步数据库操作

import asyncio
import asyncpg

async def main():
    # 连接数据库
    conn = await asyncpg.connect(
        host='localhost',
        port=5432,
        user='postgres',
        password='password',
        database='test'
    )
    
    # 执行查询
    rows = await conn.fetch('SELECT * FROM users')
    for row in rows:
        print(row)
    
    # 关闭连接
    await conn.close()

# 运行主协程
asyncio.run(main())