Python异步编程完全指南

字数统计: 1.6k字   |   阅读时长≈ 7分

Python异步编程完全指南：从原理到实战

引言：为什么需要异步编程？

在现代Web应用和分布式系统中，I/O密集型操作（如网络请求、数据库查询、文件读写）常常成为性能瓶颈。传统的同步编程模型在处理这些操作时，会阻塞整个线程，导致CPU资源闲置，系统吞吐量下降。

想象一个Web服务器需要同时处理数千个客户端请求。如果使用同步方式，每个请求都会阻塞线程直到I/O完成，服务器需要创建大量线程来维持并发，这带来了巨大的内存开销和上下文切换成本。

异步编程通过非阻塞I/O和事件循环机制，允许单个线程同时处理多个任务，显著提高了I/O密集型应用的性能和资源利用率。Python通过asyncio库提供了完整的异步编程支持，让开发者能够编写高效、可扩展的并发代码。

技术原理详解

核心概念解析

1. 事件循环（Event Loop）

事件循环是异步编程的核心引擎，负责调度和执行异步任务。它不断检查任务队列，执行就绪的任务，并在任务等待I/O时挂起，让出控制权给其他任务。

1
2
3
4
5
6
7
import asyncio

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

# 运行异步任务
loop.run_until_complete(main())

2. 协程（Coroutine）

协程是异步编程的基本执行单元，使用async def定义。协程可以在执行过程中暂停（使用await），并在适当时候恢复执行，而不阻塞整个线程。

1
2
3
4
async def fetch_data():
    # 模拟I/O操作
    await asyncio.sleep(1)
    return {"data": "example"}

3. Future和Task

  • Future：表示异步操作的最终结果，是协程的底层抽象
  • Task：是Future的子类，用于包装和管理协程的执行

异步与并发的区别

特性 异步编程 多线程/多进程
执行模型 单线程，协作式多任务 多线程/进程，抢占式多任务
上下文切换 轻量级，由程序控制 重量级，由操作系统调度
内存开销 低 高（每个线程~8MB栈空间）
适用场景 I/O密集型 CPU密集型
数据共享 安全（单线程） 需要同步机制

实战代码示例

示例1：基础异步HTTP客户端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import asyncio
import aiohttp
import time

async def fetch_url(session, url):
    """异步获取URL内容"""
    try:
        async with session.get(url, timeout=10) as response:
            data = await response.text()
            return f"URL: {url}, Status: {response.status}, Length: {len(data)}"
    except Exception as e:
        return f"URL: {url}, Error: {str(e)}"

async def main():
    urls = [
        'https://httpbin.org/get',
        'https://api.github.com',
        'https://docs.python.org',
        'https://example.com'
    ]
    
    # 创建aiohttp会话
    async with aiohttp.ClientSession() as session:
        # 创建任务列表
        tasks = [fetch_url(session, url) for url in urls]
        
        # 并发执行所有任务
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 输出结果
        for result in results:
            print(result)

# 性能对比：同步 vs 异步
async def benchmark():
    urls = ['https://httpbin.org/delay/1'] * 10
    
    # 异步版本
    start = time.time()
    await main()
    async_time = time.time() - start
    
    print(f"\n异步执行时间: {async_time:.2f}秒")
    print(f"同步预估时间: {len(urls) * 1:.2f}秒（假设每个请求1秒）")

if __name__ == "__main__":
    asyncio.run(benchmark())

示例2：生产者-消费者模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
import asyncio
import random
from collections import deque

class AsyncQueue:
    """异步队列实现"""
    def __init__(self, maxsize=0):
        self._queue = deque()
        self._maxsize = maxsize
        self._getters = deque()
        self._putters = deque()
        self._unfinished_tasks = 0
        
    async def put(self, item):
        """异步放入元素"""
        while self._maxsize and len(self._queue) >= self._maxsize:
            # 队列已满，等待消费者
            future = asyncio.Future()
            self._putters.append(future)
            await future
            
        self._queue.append(item)
        self._unfinished_tasks += 1
        
        # 唤醒等待的消费者
        if self._getters:
            getter = self._getters.popleft()
            getter.set_result(None)
            
    async def get(self):
        """异步获取元素"""
        while not self._queue:
            # 队列为空，等待生产者
            future = asyncio.Future()
            self._getters.append(future)
            await future
            
        item = self._queue.popleft()
        
        # 唤醒等待的生产者
        if self._putters:
            putter = self._putters.popleft()
            putter.set_result(None)
            
        return item
    
    def task_done(self):
        """标记任务完成"""
        self._unfinished_tasks -= 1
        
    async def join(self):
        """等待所有任务完成"""
        while self._unfinished_tasks:
            await asyncio.sleep(0.1)

async def producer(queue, producer_id, count=5):
    """生产者协程"""
    for i in range(count):
        item = f"产品-{producer_id}-{i}"
        await asyncio.sleep(random.uniform(0.1, 0.5))  # 模拟生产时间
        await queue.put(item)
        print(f"生产者 {producer_id} 生产了: {item}")
    print(f"生产者 {producer_id} 完成")

async def consumer(queue, consumer_id):
    """消费者协程"""
    while True:
        try:
            item = await asyncio.wait_for(queue.get(), timeout=2)
            await asyncio.sleep(random.uniform(0.2, 0.8))  # 模拟消费时间
            print(f"消费者 {consumer_id} 消费了: {item}")
            queue.task_done()
        except asyncio.TimeoutError:
            print(f"消费者 {consumer_id} 超时退出")
            break

async def run_producer_consumer():
    """运行生产者-消费者模式"""
    queue = AsyncQueue(maxsize=3)
    
    # 创建生产者和消费者任务
    producers = [producer(queue, i) for i in range(2)]
    consumers = [consumer(queue, i) for i in range(3)]
    
    # 启动所有任务
    all_tasks = producers + consumers
    await asyncio.gather(*all_tasks)
    
    # 等待队列清空
    await queue.join()

if __name__ == "__main__":
    asyncio.run(run_producer_consumer())

示例3：异步上下文管理器与错误处理

import asyncio
import aiohttp
from contextlib import asynccontextmanager

class AsyncDatabase:
    """模拟异步数据库连接"""
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.is_connected = False
        
    async def connect(self):
        """模拟连接数据库"""
        await asyncio.sleep(0.5)
        self.is_connected = True
        print(f"已连接到数据库: {self.connection_string}")
        
    async def disconnect(self):
        """模拟断开数据库连接"""
        await asyncio.sleep(0.2)
        self.is_connected = False
        print("数据库连接已关闭")
        
    async def query(self, sql):
        """模拟数据库查询"""
        if not self.is_connected:
            raise ConnectionError("数据库未连接")
        
        await asyncio.sleep(0.3)  # 模拟查询时间
        return f"查询结果: {sql}"

@asynccontextmanager
async def database_session(connection_string):
    """异步上下文管理器"""
    db = AsyncDatabase(connection_string)
    try:
        await db.connect()
        yield db
    finally:
        await db.disconnect()

async def execute_transaction(db, queries):
    """执行事务操作"""
    results = []
    for query in queries:
        try:
            result = await db.query(query)
            results.append(result)
        except Exception as e:
            print(f"查询失败: {query}, 错误: {e}")
            # 这里可以添加回滚逻辑
            raise
    return results

async def main_with_context():
    """使用异步上下文管理器"""
    connection_string = "postgresql://user:pass

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发、聚合支付、
开源爱好者、Linux

联系QQ:2067409116

很惭愧

只做了一点微小的工作
谢谢大家
切换形象
转人工
🔮 今日运势
✨ 切换至 Live2D
收起
客服系统
🗑️ ✕

确定要重置当前对话吗？

确定
取消