Python asyncio 完全指南 | 极客日志

Python

Python asyncio 完全指南

Python asyncio 异步 I/O 框架。涵盖核心概念如协程、事件循环、Task 和 Future。对比了 asyncio 与多线程、多进程的区别，指出其在 I/O 密集型任务中的优势。包含基础用法、高级特性（上下文管理器、迭代器、队列）及实战示例（爬虫、数据库、Web 服务器）。最后总结了最佳实践，强调避免阻塞调用和合理错误处理。适合需要高并发网络应用的开发者学习。

暗影行者发布于 2026/3/26更新于 2026/4/183 浏览

什么是 asyncio

asyncio 是 Python 3.4+ 引入的标准库，用于编写并发代码的异步 I/O 框架。它使用事件循环和协程实现单线程并发，特别适合处理 I/O 密集型任务。

核心优势:

单线程内实现高并发
避免线程切换开销
更少的内存占用
代码更易于理解和调试

核心概念

协程 (Coroutine)

使用 async def 定义的特殊函数，可以在执行过程中暂停和恢复。

async def ():
    ()
     asyncio.sleep()
    ()

import asyncio

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

# 方式 1: Python 3.7+ 推荐
asyncio.run(hello())

# 方式 2: 手动管理事件循环
loop = asyncio.get_event_loop()
loop.run_until_complete(hello())
loop.close()

# 方式 3: 在已有事件循环中
# await hello() # 只能在异步函数内使用

async def task1():
    await asyncio.sleep(2)
    return "任务 1 完成"

async def task2():
    await asyncio.sleep(1)
    return "任务 2 完成"

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

asyncio.run(main())

async def fetch_data(n):
    print(f"开始获取数据 {n}")
    await asyncio.sleep(1)
    print(f"完成获取数据 {n}")
    return f"数据 {n}"

async def main():
    # gather: 并发执行，按顺序返回结果
    results = await asyncio.gather(
        fetch_data(1), fetch_data(2), fetch_data(3)
    )
    print(results)

asyncio.run(main())

┌─────────────────────────────────┐
│ 事件循环 (Event Loop)           │
│                                 │
│ ┌──────────────────────────┐   │
│ │ 就绪队列 (Ready Queue)   │   │
│ │ [task1, task2, task3]    │   │
│ └──────────────────────────┘   │
│                                 │
│ ┌──────────────────────────┐   │
│ │ 等待队列 (Wait Queue)    │   │
│ │ [task4, task5]           │   │
│ └──────────────────────────┘   │
│                                 │
│ ┌──────────────────────────┐   │
│ │ I/O 选择器                │   │
│ │ (epoll/kqueue/select)    │   │
│ └──────────────────────────┘   │
└─────────────────────────────────┘

async def example():
    print("1. 开始执行")
    # RUNNING
    await asyncio.sleep(1)
    # WAITING (挂起)
    print("2. 继续执行")
    # RUNNING (恢复)
    return "完成"
    # FINISHED

# 简化的事件循环伪代码
class EventLoop:
    def __init__(self):
        self._ready = deque()  # 就绪队列
        self._selector = select.epoll()  # I/O 选择器

    def run_until_complete(self, coro):
        task = Task(coro)
        self._ready.append(task)
        while self._ready or self._has_pending_io():
            # 执行就绪的任务
            if self._ready:
                task = self._ready.popleft()
                task.step()
            # 等待 I/O 事件
            events = self._selector.select(timeout=0)
            for event in events:
                self._ready.append(event.callback)

特性	asyncio	多线程 (threading)	多进程 (multiprocessing)
并发模型	协作式并发	抢占式并发	真正并行
运行环境	单线程	多线程	多进程
GIL 影响	无影响	受 GIL 限制	不受 GIL 限制
切换开销	极小 (用户态)	较大 (内核态)	最大 (进程切换)
内存占用	低	中等	高
适用场景	I/O 密集型	I/O 密集型	CPU 密集型
数据共享	简单 (同一线程)	需要锁	需要 IPC
调试难度	容易	困难	中等

# 典型的 asyncio 场景：并发网络请求
import aiohttp
import asyncio

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

async def main():
    urls = ['http://example.com'] * 100
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        print(f"完成 {len(results)} 个请求")
    asyncio.run(main())

import threading
import requests

def fetch_url(url):
    response = requests.get(url)
    print(f"完成：{url}")

urls = ['http://example.com'] * 10
threads = [threading.Thread(target=fetch_url, args=(url,)) for url in urls]
for t in threads:
    t.start()
for t in threads:
    t.join()

from multiprocessing import Pool

def cpu_intensive(n):
    return sum(i*i for i in range(n))

with Pool(4) as pool:
    results = pool.map(cpu_intensive, [10000000]*4)
    print(results)

import asyncio
import threading
import multiprocessing
import time

# I/O 密集型任务
def io_bound_sync():
    time.sleep(1)

async def io_bound_async():
    await asyncio.sleep(1)

# 测试 asyncio
async def test_asyncio():
    start = time.time()
    await asyncio.gather(*[io_bound_async() for _ in range(100)])
    print(f"asyncio: {time.time() - start:.2f}s")  # 约 1 秒

# 测试多线程
def test_threading():
    start = time.time()
    threads = [threading.Thread(target=io_bound_sync) for _ in range(100)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(f"threading: {time.time() - start:.2f}s")  # 约 1-2 秒

# asyncio 在 I/O 密集型任务中表现最优

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

async def main():
    async with AsyncResource() as resource:
        print("使用资源")
    asyncio.run(main())

class AsyncRange:
    def __init__(self, start, end):
        self.current = start
        self.end = end

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.current >= self.end:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)
        self.current += 1
        return self.current - 1

async def main():
    async for i in AsyncRange(0, 5):
        print(i)
    asyncio.run(main())

async def async_generator():
    for i in range(5):
        await asyncio.sleep(1)
        yield i

async def main():
    async for value in async_generator():
        print(value)
    asyncio.run(main())

async def slow_operation():
    await asyncio.sleep(5)
    return "完成"

async def main():
    try:
        # 设置 2 秒超时
        result = await asyncio.wait_for(slow_operation(), timeout=2)
    except asyncio.TimeoutError:
        print("操作超时")
    asyncio.run(main())

# 限制并发数量
async def limited_task(sem, n):
    async with sem:
        print(f"任务 {n} 开始")
        await asyncio.sleep(1)
        print(f"任务 {n} 完成")

async def main():
    sem = asyncio.Semaphore(3)  # 最多 3 个并发
    await asyncio.gather(*[limited_task(sem, i) for i in range(10)])
    asyncio.run(main())

async def producer(queue, n):
    for i in range(n):
        await queue.put(i)
        print(f"生产：{i}")
    await asyncio.sleep(0.5)

async def consumer(queue, name):
    while True:
        item = await queue.get()
        print(f"{name} 消费：{item}")
        await asyncio.sleep(1)
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    # 创建生产者和消费者
    producers = [asyncio.create_task(producer(queue, 5))]
    consumers = [asyncio.create_task(consumer(queue, f"消费者{i}")) for i in range(2)]
    await asyncio.gather(*producers)
    await queue.join()  # 等待所有任务处理完成
    for c in consumers:
        c.cancel()
    asyncio.run(main())

import asyncio
import aiohttp
from bs4 import BeautifulSoup

async def fetch_page(session, url):
    try:
        async with session.get(url, timeout=10) as response:
            return await response.text()
    except Exception as e:
        print(f"错误 {url}: {e}")
    return None

async def parse_page(html):
    if html:
        soup = BeautifulSoup(html, 'html.parser')
        return soup.title.string if soup.title else "无标题"
    return None

async def crawl_urls(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_page(session, url) for url in urls]
        pages = await asyncio.gather(*tasks)
        titles = await asyncio.gather(*[parse_page(page) for page in pages])
        return titles

urls = ['http://example.com', 'http://example.org', 'http://example.net']
# 执行爬虫
# titles = asyncio.run(crawl_urls(urls))
# print(titles)

import asyncio
import aiosqlite

async def create_table(db):
    await db.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY,
            name TEXT,
            email TEXT
        )
    ''')
    await db.commit()

async def insert_user(db, name, email):
    await db.execute('INSERT INTO users (name, email) VALUES (?, ?)', (name, email))
    await db.commit()

async def fetch_users(db):
    async with db.execute('SELECT * FROM users') as cursor:
        return await cursor.fetchall()

async def main():
    async with aiosqlite.connect('test.db') as db:
        await create_table(db)
        # 并发插入
        await asyncio.gather(
            insert_user(db, 'Alice', '[email protected]'),
            insert_user(db, 'Bob', '[email protected]'),
            insert_user(db, 'Charlie', '[email protected]')
        )
        users = await fetch_users(db)
        for user in users:
            print(user)
    # asyncio.run(main())

import asyncio

async def handle_client(reader, writer):
    data = await reader.read(1024)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    print(f"收到来自 {addr} 的消息：{message}")
    response = f"Echo: {message}"
    writer.write(response.encode())
    await writer.drain()
    writer.close()
    await writer.wait_closed()

async def start_server():
    server = await asyncio.start_server(
        handle_client, '127.0.0.1', 8888
    )
    addr = server.sockets[0].getsockname()
    print(f"服务器启动在 {addr}")
    async with server:
        await server.serve_forever()
# asyncio.run(start_server())

import asyncio
import random

async def data_stream():
    """模拟数据流"""
    while True:
        yield random.randint(1, 100)
        await asyncio.sleep(0.5)

async def process_data(value):
    """处理数据"""
    await asyncio.sleep(0.1)
    return value * 2

async def monitor_stream():
    """监控和处理数据流"""
    buffer = []
    async for data in data_stream():
        print(f"接收数据：{data}")
        # 异步处理数据
        task = asyncio.create_task(process_data(data))
        buffer.append(task)
        # 每 5 个数据批量处理
        if len(buffer) >= 5:
            results = await asyncio.gather(*buffer)
            print(f"处理结果：{results}")
            buffer = []
        # 演示用，处理 20 个数据后停止
        if data > 20:
            break
    # asyncio.run(monitor_stream())

import asyncio

async def task_with_priority(name, priority, duration):
    print(f"[优先级 {priority}] {name} 开始")
    await asyncio.sleep(duration)
    print(f"[优先级 {priority}] {name} 完成")
    return f"{name} 结果"

async def coordinator():
    # 创建不同优先级的任务
    high_priority = [
        task_with_priority(f"高优先级-{i}", 1, 1)
        for i in range(3)
    ]
    low_priority = [
        task_with_priority(f"低优先级-{i}", 3, 2)
        for i in range(3)
    ]
    # 先执行高优先级任务
    high_results = await asyncio.gather(*high_priority)
    print(f"高优先级完成：{high_results}")
    # 再执行低优先级任务
    low_results = await asyncio.gather(*low_priority)
    print(f"低优先级完成：{low_results}")
    asyncio.run(coordinator())

async def safe_task(n):
    try:
        if n == 3:
            raise ValueError("错误的值")
        await asyncio.sleep(1)
        return f"任务 {n} 成功"
    except Exception as e:
        print(f"任务 {n} 失败：{e}")
        return None

async def main():
    results = await asyncio.gather(
        safe_task(1),
        safe_task(2),
        safe_task(3),
        return_exceptions=True  # 不会因为一个任务失败而停止
    )
    print(results)
    asyncio.run(main())

class AsyncConnection:
    async def __aenter__(self):
        self.conn = await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()

    async def connect(self):
        print("建立连接")
        await asyncio.sleep(1)
        return "connection"

    async def close(self):
        print("关闭连接")
        await asyncio.sleep(1)

async def main():
    async with AsyncConnection() as conn:
        print(f"使用连接：{conn.conn}")
    asyncio.run(main())

import asyncio
from concurrent.futures import ThreadPoolExecutor

# 错误：阻塞操作
async def bad_example():
    time.sleep(5)  # 这会阻塞整个事件循环!

# 正确：使用 executor 运行阻塞操作
async def good_example():
    loop = asyncio.get_event_loop()
    with ThreadPoolExecutor() as executor:
        result = await loop.run_in_executor(executor, blocking_function)
    return result

def blocking_function():
    import time
    time.sleep(5)
    return "完成"

async def with_timeout():
    try:
        result = await asyncio.wait_for(
            long_running_task(), timeout=5.0
        )
    except asyncio.TimeoutError:
        print("任务超时，执行回退逻辑")
        result = "默认值"
    return result

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(some_coro())
        task2 = tg.create_task(another_coro())
    # 所有任务完成或某个任务失败时退出
    print("所有任务完成")

Python asyncio 完全指南

什么是 asyncio

核心概念

协程 (Coroutine)

事件循环 (Event Loop)

await 关键字

Task

Future

基础用法

1. 运行协程的三种方式

2. 创建和管理任务

3. 并发执行

内部机理

事件循环的工作原理

执行流程

协程状态转换

底层实现关键点

与多线程/多进程的区别

对比表格

使用场景详解

asyncio 适合的场景

多线程适合的场景

多进程适合的场景

性能对比示例

高级特性

1. 异步上下文管理器

2. 异步迭代器

3. 异步生成器

4. 超时控制

5. 信号量和锁

6. 队列

实战示例

示例 1: 异步网络爬虫

示例 2: 异步数据库操作

示例 3: 异步 Web 服务器

示例 4: 实时数据流处理

示例 5: 多任务协调

最佳实践

1. 错误处理

2. 资源清理

3. 避免阻塞

4. 合理设置超时

5. 使用 TaskGroup (Python 3.11+)

总结

asyncio 的核心优势

何时使用 asyncio

学习路径建议

常见陷阱

参考资源

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具