Python 异步编程与协程实战指南 | 极客日志

Python大前端算法

Python 异步编程与协程实战指南

Python 异步编程通过非阻塞 I/O 提升效率。核心涉及协程、事件循环及 asyncio 库。 async/await 语法，展示任务调度、超时处理及取消机制。结合 aiohttp 实现并发 HTTP 请求与 Web 服务器搭建，提供高并发场景下的实战代码示例，帮助开发者掌握高性能网络应用开发技巧。

监控大屏发布于 2026/3/23更新于 2026/4/284 浏览

Python 异步编程与协程实战指南

在 Python 开发中，面对高并发的 I/O 密集型任务，传统的同步阻塞模式往往成为性能瓶颈。异步编程通过非阻塞操作，让程序在等待网络或磁盘响应时能继续处理其他逻辑，从而显著提升执行效率。本文将深入探讨协程、事件循环及 asyncio 库的核心机制，并通过 aiohttp 实战案例展示如何构建高性能的异步应用。

异步编程核心概念

什么是异步编程

异步编程允许程序在执行耗时操作（如 HTTP 请求、文件读写）时不阻塞主线程。当遇到 I/O 等待时，控制权会交还给事件循环，去调度其他就绪的任务。这种方式不仅减少了线程切换开销，还简化了并发代码的结构。

典型应用场景

网络通信：HTTP 请求、WebSocket 长连接、微服务调用。
文件操作：大文件的流式读取与写入。
数据库交互：异步 ORM 查询，避免阻塞数据库连接池。

协程的定义与使用

协程（Coroutine）是轻量级的用户态线程，支持暂停和恢复。在 Python 中，使用 async def 定义协程函数，内部通过 await 挂起执行。

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():
    # gather 用于并发执行多个协程
    await asyncio.gather(count(), count(), count())

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

async def hello():
    print('Hello, World!')
    await asyncio.sleep(1)
    print('Hello again!')

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())

pip install aiohttp

import aiohttp
import asyncio

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())

import aiohttp
import asyncio
import 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())

import aiohttp
import asyncio
import time

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

async def main():
    urls = [
        'https://www.example.com',
        'https://www.google.com',
        'https://www.github.com',
        'https://www.python.org'
    ]
    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}秒')

from aiohttp import web
import asyncio

async def handle_root(request):
    return web.Response(text='Hello, World!')

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')

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

if __name__ == '__main__':
    asyncio.run(web.run_app(create_app(), host='localhost', port=8080))