Python异步IO之asyncio举例深度解析_Python

1. 异步编程概述

异步编程是一种编程范式，它允许程序在等待某个操作完成时继续执行其他任务，而不是阻塞等待。在python中，asyncio库是实现异步编程的核心。

核心概念

同步编程：代码按顺序执行，一个操作完成后才开始下一个操作
异步编程：代码可以在等待某些操作（如i/o）时执行其他任务
协程：轻量级的并发执行单元
事件循环：管理和调度协程的执行
future：表示异步操作的结果
task：对协程的封装，用于并发执行

2. asyncio基础

2.1 协程定义

在python中，协程可以通过async def关键字定义：

import asyncio

async def hello():
    print('hello')
    await asyncio.sleep(1)
    print('world')

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

2.2 事件循环

事件循环是asyncio的核心，它负责调度协程的执行：

import asyncio

async def task1():
    print('task 1 started')
    await asyncio.sleep(2)
    print('task 1 completed')

async def task2():
    print('task 2 started')
    await asyncio.sleep(1)
    print('task 2 completed')

async def main():
    # 创建任务
    t1 = asyncio.create_task(task1())
    t2 = asyncio.create_task(task2())
    
    # 等待任务完成
    await t1
    await t2

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

2.3 await关键字

await关键字用于等待一个异步操作完成：

async def fetch_data(url):
    print(f'fetching data from {url}')
    # 模拟网络请求
    await asyncio.sleep(2)
    return f'data from {url}'

async def main():
    # 串行执行
    data1 = await fetch_data('https://api.example.com/data1')
    data2 = await fetch_data('https://api.example.com/data2')
    print(data1, data2)

asyncio.run(main())

3. 并发执行

3.1 asyncio.gather

asyncio.gather用于并发执行多个协程：

async def fetch_data(url):
    print(f'fetching data from {url}')
    await asyncio.sleep(2)
    return f'data from {url}'

async def main():
    # 并发执行
    results = await asyncio.gather(
        fetch_data('https://api.example.com/data1'),
        fetch_data('https://api.example.com/data2'),
        fetch_data('https://api.example.com/data3')
    )
    print(results)

asyncio.run(main())

3.2 asyncio.create_task

asyncio.create_task用于创建后台任务：

async def background_task():
    while true:
        print('background task running')
        await asyncio.sleep(1)

async def main():
    # 创建后台任务
    task = asyncio.create_task(background_task())
    
    # 执行其他操作
    print('main task running')
    await asyncio.sleep(3)
    
    # 取消后台任务
    task.cancel()
    try:
        await task
    except asyncio.cancellederror:
        print('background task cancelled')

asyncio.run(main())

3.3 asyncio.wait

asyncio.wait用于等待多个协程完成：

async def task1():
    await asyncio.sleep(2)
    return 'task 1 result'

async def task2():
    await asyncio.sleep(1)
    return 'task 2 result'

async def main():
    tasks = [task1(), task2()]
    done, pending = await asyncio.wait(tasks, timeout=1.5)
    
    print('done tasks:', len(done))
    print('pending tasks:', len(pending))
    
    for task in done:
        print('result:', await task)

asyncio.run(main())

4. 异步io操作

4.1 文件io

使用aiofiles库进行异步文件操作：

import asyncio
import aiofiles

async def read_file(filename):
    async with aiofiles.open(filename, 'r') as f:
        content = await f.read()
    return content

async def write_file(filename, content):
    async with aiofiles.open(filename, 'w') as f:
        await f.write(content)

async def main():
    # 读取文件
    content = await read_file('example.txt')
    print('file content:', content)
    
    # 写入文件
    await write_file('output.txt', 'hello, asyncio!')
    print('file written')

asyncio.run(main())

4.2 网络io

使用aiohttp库进行异步网络请求：

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.clientsession() as session:
        html = await fetch(session, 'https://example.com')
        print('html length:', len(html))

asyncio.run(main())

4.3 数据库操作

使用asyncpg库进行异步数据库操作：

import asyncio
import asyncpg

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

asyncio.run(main())

5. 高级特性

5.1 异步上下文管理器

使用async with语句创建异步上下文管理器：

import asyncio

class asynccontextmanager:
    async def __aenter__(self):
        print('entering context')
        await asyncio.sleep(1)
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print('exiting context')
        await asyncio.sleep(1)

async def main():
    async with asynccontextmanager() as cm:
        print('inside context')
        await asyncio.sleep(2)

asyncio.run(main())

5.2 异步迭代器

使用async for语句创建异步迭代器：

import asyncio

class asynciterator:
    def __init__(self, start, end):
        self.start = start
        self.end = end
    
    def __aiter__(self):
        self.current = self.start
        return self
    
    async def __anext__(self):
        if self.current >= self.end:
            raise stopasynciteration
        value = self.current
        self.current += 1
        await asyncio.sleep(0.5)
        return value

async def main():
    async for num in asynciterator(1, 5):
        print(num)

asyncio.run(main())

5.3 任务组

python 3.11+引入了任务组，用于更安全地管理并发任务：

import asyncio

async def task(id, duration):
    print(f'task {id} started')
    await asyncio.sleep(duration)
    print(f'task {id} completed')
    return f'task {id} result'

async def main():
    async with asyncio.taskgroup() as tg:
        # 创建任务
        task1 = tg.create_task(task(1, 2))
        task2 = tg.create_task(task(2, 1))
        task3 = tg.create_task(task(3, 3))
    
    # 任务组退出时，所有任务已完成
    print('all tasks completed')
    print('task 1 result:', task1.result())
    print('task 2 result:', task2.result())
    print('task 3 result:', task3.result())

asyncio.run(main())

6. 性能优化

6.1 避免阻塞操作

在异步代码中避免使用阻塞操作，如同步io：

# 错误示例
async def bad_example():
    # 阻塞操作
    time.sleep(1)  # 这会阻塞整个事件循环
    print('done')

# 正确示例
async def good_example():
    # 异步操作
    await asyncio.sleep(1)  # 这会释放事件循环
    print('done')

6.2 批量操作

对于多个io操作，使用并发执行：

async def fetch_all(urls):
    async with aiohttp.clientsession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    return results

6.3 超时处理

为异步操作设置超时：

async def fetch_with_timeout(url, timeout=5):
    try:
        async with aiohttp.clientsession() as session:
            async with asyncio.timeout(timeout):
                async with session.get(url) as response:
                    return await response.text()
    except asyncio.timeouterror:
        return 'request timed out'

7. 常见陷阱

7.1 忘记await

忘记使用await会导致协程不会执行：

async def foo():
    print('foo')
    await asyncio.sleep(1)
    print('bar')

async def main():
    foo()  # 错误：忘记await，协程不会执行
    await foo()  # 正确：使用await

asyncio.run(main())

7.2 阻塞事件循环

在协程中使用阻塞操作会阻塞整个事件循环：

async def blocking_operation():
    # 错误：使用阻塞操作
    time.sleep(1)  # 这会阻塞事件循环
    return 'done'

async def main():
    # 正确：使用线程池执行阻塞操作
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(none, lambda: time.sleep(1))
    return result

7.3 任务泄漏

创建的任务如果不等待或取消，会导致任务泄漏：

async def background_task():
    while true:
        await asyncio.sleep(1)
        print('background task')

async def main():
    # 错误：创建任务但不管理
    asyncio.create_task(background_task())
    await asyncio.sleep(5)
    # 任务会继续运行，导致泄漏

async def main_fixed():
    # 正确：管理任务生命周期
    task = asyncio.create_task(background_task())
    await asyncio.sleep(5)
    task.cancel()
    try:
        await task
    except asyncio.cancellederror:
        pass

8. 代码示例

8.1 异步web服务器

使用aiohttp创建异步web服务器：

from aiohttp import web

async def handle(request):
    name = request.match_info.get('name', 'world')
    # 模拟异步操作
    await asyncio.sleep(0.5)
    return web.response(text=f'hello, {name}!')

async def main():
    app = web.application()
    app.add_routes([
        web.get('/', handle),
        web.get('/{name}', handle)
    ])
    runner = web.apprunner(app)
    await runner.setup()
    site = web.tcpsite(runner, 'localhost', 8080)
    await site.start()
    print('server started on http://localhost:8080')
    # 保持运行
    await asyncio.future()  # 无限等待

if __name__ == '__main__':
    asyncio.run(main())

8.2 异步爬虫

使用aiohttp创建异步爬虫：

import asyncio
import aiohttp
from bs4 import beautifulsoup

async def fetch_url(session, url):
    try:
        async with session.get(url) as response:
            return await response.text()
    except exception as e:
        print(f'error fetching {url}: {e}')
        return ''

async def parse_page(html):
    soup = beautifulsoup(html, 'html.parser')
    links = []
    for a in soup.find_all('a', href=true):
        links.append(a['href'])
    return links

async def crawl(start_url, max_depth=2):
    visited = set()
    queue = [(start_url, 0)]
    
    async with aiohttp.clientsession() as session:
        while queue:
            url, depth = queue.pop(0)
            if url in visited or depth >= max_depth:
                continue
            
            visited.add(url)
            print(f'crawling {url} (depth: {depth})')
            
            html = await fetch_url(session, url)
            if not html:
                continue
            
            links = await parse_page(html)
            for link in links:
                if link.startswith('http'):
                    queue.append((link, depth + 1))

async def main():
    await crawl('https://example.com')

asyncio.run(main())

8.3 异步数据库操作

使用asyncpg进行异步数据库操作：

import asyncio
import asyncpg

async def setup_database():
    # 连接数据库
    conn = await asyncpg.connect(
        host='localhost',
        port=5432,
        user='postgres',
        password='password',
        database='mydb'
    )
    
    # 创建表
    await conn.execute('''
        create table if not exists users (
            id serial primary key,
            name varchar(100),
            email varchar(100) unique
        )
    ''')
    
    # 插入数据
    await conn.execute(
        'insert into users (name, email) values ($1, $2) on conflict do nothing',
        'alice', 'alice@example.com'
    )
    await conn.execute(
        'insert into users (name, email) values ($1, $2) on conflict do nothing',
        'bob', 'bob@example.com'
    )
    
    # 查询数据
    rows = await conn.fetch('select * from users')
    print('users:')
    for row in rows:
        print(f'id: {row["id"]}, name: {row["name"]}, email: {row["email"]}')
    
    # 关闭连接
    await conn.close()

asyncio.run(setup_database())

9. 工具与生态

9.1 常用库

aiohttp：异步http客户端/服务器
aiofiles：异步文件操作
asyncpg：异步postgresql客户端
motor：异步mongodb客户端
redis-py：支持异步的redis客户端
httpx：现代http客户端，支持同步和异步

9.2 开发工具

asyncio-debug：调试异步代码
aiomonitor：监控异步应用
tqdm：支持异步的进度条

9.3 框架

fastapi：现代异步web框架
sanic：高性能异步web框架
tornado：异步web框架
quart：flask的异步版本

10. 结论

asyncio为python带来了强大的异步编程能力，使得我们可以编写高效的i/o密集型应用。通过合理使用asyncio的各种特性，我们可以：

提高应用的并发性能
减少资源消耗
编写更清晰、更可维护的代码
充分利用现代硬件的多核性能

最佳实践

使用异步库：尽量使用支持异步的库，如aiohttp、aiofiles等
避免阻塞操作：在异步代码中避免使用阻塞操作
合理使用并发：使用asyncio.gather等工具并发执行任务
管理任务生命周期：确保任务被正确等待或取消
设置超时：为异步操作设置合理的超时
使用任务组：在python 3.11+中使用taskgroup管理任务
监控和调试：使用适当的工具监控和调试异步代码

未来发展

更广泛的异步支持：越来越多的库将支持异步操作
更简洁的语法：python可能会引入更简洁的异步语法
更好的工具支持：更多专门用于异步编程的工具和框架
更深入的集成：异步编程将与python生态系统更深入集成

通过掌握asyncio，我们可以构建高性能、可扩展的python应用，特别是在处理大量i/o操作的场景中，如web服务器、爬虫、数据处理等。

到此这篇关于python异步io之asyncio深度解析的文章就介绍到这了,更多相关python异步io之asyncio内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！

Python异步IO之asyncio举例深度解析