Python 异步编程与协程详解

一、学习目标与重点

学习目标：掌握 Python 异步编程的基本概念和方法，包括协程、任务调度、事件循环等；学习 asyncio、aiohttp 等核心库的使用；通过实战案例开发异步应用程序。

学习重点：协程的定义与使用、任务调度、事件循环、asyncio 库、aiohttp 库、异步编程实战。

二、异步编程概述

什么是异步编程

异步编程是一种并发编程方式，通过非阻塞的操作提高程序的执行效率。在异步编程中，程序可以在等待 I/O 操作完成时继续执行其他任务，而不需要阻塞等待。

异步编程的优势

提高执行效率：在等待 I/O 操作完成时，程序可以继续执行其他任务。
降低资源消耗：减少了线程切换的开销。
简化代码结构：通过协程和任务调度，代码结构更加简洁。

异步编程的应用场景

网络通信：如 HTTP 请求、Web 服务器、WebSocket 通信等。
文件操作：如大文件的读取和写入。
数据库操作：如异步数据库查询。

三、协程的定义与使用

协程的定义

协程（Coroutine）是一种轻量级的线程，可以在程序中进行暂停和恢复。在 Python 中，协程可以通过 async def 关键字定义。

协程的使用

import asyncio

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

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

协程的暂停与恢复

import asyncio

# 定义协程
async def count():
    print('Counting...')
    await asyncio.sleep(1)
    print('Counted!')

# 运行多个协程
async def main():
    await asyncio.gather(count(), count(), count())

asyncio.run(main())

四、任务调度

创建任务

import asyncio

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

# 创建任务
async def main():
    task1 = asyncio.create_task(hello())
    task2 = asyncio.create_task(hello())
    await task1
    await task2

asyncio.run(main())

任务的取消

import asyncio

# 定义协程
async def hello():
    try:
        print('Hello, World!')
        await asyncio.sleep(1)
        print('Hello again!')
    except asyncio.CancelledError:
        print('Task cancelled!')

# 创建任务并取消
async def main():
    task = asyncio.create_task(hello())
    await asyncio.sleep(0.5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print('Main: Task cancelled!')

asyncio.run(main())

任务的超时

import asyncio

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(2)
    print('Hello again!')

# 任务超时
async def main():
    try:
        await asyncio.wait_for(hello(), timeout=1)
    except asyncio.TimeoutError:
        print('Task timed out!')

asyncio.run(main())

五、事件循环

事件循环的概述

事件循环是异步编程的核心组件，负责调度任务的执行。事件循环会不断地从任务队列中取出任务并执行，直到任务队列为空。

获取事件循环

import asyncio

# 获取事件循环
loop = asyncio.get_event_loop()

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

# 运行协程
loop.run_until_complete(hello())

事件循环的运行

import asyncio

# 获取事件循环
loop = asyncio.get_event_loop()

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

# 运行多个协程
loop.run_until_complete(asyncio.gather(hello(), hello(), hello()))

六、asyncio 库

asyncio 的基本用法

import asyncio

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

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

asyncio 的常用函数

asyncio.run()：运行协程。
asyncio.create_task()：创建任务。
asyncio.gather()：收集多个协程的结果。
asyncio.wait_for()：等待协程完成，设置超时。
asyncio.sleep()：暂停协程。

asyncio 的高级用法

import asyncio

# 定义协程
async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

# 使用 Future 对象
async def main():
    future = asyncio.Future()
    task = asyncio.create_task(hello())
    task.add_done_callback(lambda t: future.set_result(t.result()))
    await future

asyncio.run(main())

七、aiohttp 库

安装 aiohttp

pip install aiohttp

发送 HTTP 请求

import aiohttp
import asyncio

# 发送 GET 请求
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://www.example.com')
        print(html)

asyncio.run(main())

发送 POST 请求

import aiohttp
import asyncio
import json

# 发送 POST 请求
async def post_data(session, url, data):
    async with session.post(url, data=data) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        data = {'name': '张三', 'age': 25}
        response = await post_data(session, 'https://httpbin.org/post', data)
        print(response)

asyncio.run(main())

发送 JSON 请求

import aiohttp
import asyncio
import json

# 发送 JSON 请求
async def post_json(session, url, data):
    async with session.post(url, json=data) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        data = {'name': '张三', 'age': 25}
        response = await post_json(session, 'https://httpbin.org/post', data)
        print(response)

asyncio.run(main())

八、实战案例：异步 HTTP 客户端

需求分析

开发一个异步 HTTP 客户端，支持以下功能：

发送 HTTP 请求。
并发发送多个 HTTP 请求。
处理响应数据。

代码实现

import aiohttp
import asyncio
import time

# 发送 HTTP 请求
async def fetch(session, url):
    start_time = time.time()
    async with session.get(url) as response:
        text = await response.text()
        elapsed_time = time.time() - start_time
        return url, len(text), elapsed_time

# 并发发送 HTTP 请求
async def main():
    urls = ['https://www.example.com', 'https://www.google.com', 'https://www.github.com', 'https://www.python.org', 'https://www.djangoproject.com']
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, url)) for url in urls]
        results = await asyncio.gather(*tasks)
        for url, length, elapsed_time in results:
            print(f'URL: {url}, 响应长度：{length}, 耗时：{elapsed_time:.2f}秒')

# 运行程序
if __name__ == '__main__':
    start_time = time.time()
    asyncio.run(main())
    elapsed_time = time.time() - start_time
    print(f'总耗时：{elapsed_time:.2f}秒')

实施过程

安装 aiohttp 库。
定义发送 HTTP 请求的协程函数。
定义并发发送 HTTP 请求的协程函数。
运行程序。

最终效果

通过异步 HTTP 客户端，可以实现以下功能：

发送 HTTP 请求。
并发发送多个 HTTP 请求。
处理响应数据。

九、实战案例：异步 Web 服务器

需求分析

开发一个异步 Web 服务器，支持以下功能：

处理 HTTP 请求。
提供静态文件服务。
实现简单的 API 接口。

代码实现

from aiohttp import web
import asyncio

# 处理根路径请求
async def handle_root(request):
    return web.Response(text='Hello, World!')

# 处理 API 接口请求
async def handle_api(request):
    data = {'name': '张三', 'age': 25}
    return web.json_response(data)

# 提供静态文件服务
async def handle_static(request):
    return web.FileResponse('static/index.html')

# 创建 Web 应用
async def create_app():
    app = web.Application()
    # 添加路由
    app.add_routes([
        web.get('/', handle_root),
        web.get('/api', handle_api),
        web.get('/static/{name}', handle_static)
    ])
    return app

# 运行 Web 服务器
if __name__ == '__main__':
    asyncio.run(web.run_app(create_app(), host='localhost', port=8080))

实施过程

安装 aiohttp 库。
定义处理 HTTP 请求的协程函数。
创建 Web 应用。
添加路由。
运行 Web 服务器。

最终效果

通过异步 Web 服务器，可以实现以下功能：

处理 HTTP 请求。
提供静态文件服务。
实现简单的 API 接口。

总结

本文涵盖了 Python 异步编程的核心概念与实践方法，包括协程、任务调度及事件循环机制。详细介绍了 asyncio 库的基础函数与高级用法，并结合 aiohttp 库演示了 HTTP 请求的发送流程。通过构建异步 HTTP 客户端与 Web 服务器的实战案例，展示了高并发场景下的代码实现与性能优势。建议读者结合代码练习以加深理解。