🔎大家好，我是ZTLJQ，希望你看完之后，能对你有所帮助，不足请指正！共同学习交流

📝个人主页－ZTLJQ的主页

🎁欢迎各位→点赞👍 + 收藏⭐️ + 留言📝​📣系列果你对这个系列感兴趣的话

专栏 - ​​​​​​Python从零到企业级应用：短时间成为市场抢手的程序员

✔说明⇢本人讲解主要包括Python爬虫、JS逆向、Python的企业级应用

如果你对这个系列感兴趣的话，可以关注订阅哟👋

告别阻塞，迎接并发

在传统的同步编程模式下，程序会逐行执行，当遇到一个耗时的操作（如网络请求、文件读写、数据库查询）时，整个线程会陷入等待，无法处理其他任务。这种“阻塞”现象在处理大量I/O密集型任务时，会严重浪费CPU资源，导致应用性能低下。

Python的异步编程为我们打开了一扇通往高并发世界的大门。通过asyncio库和async/await关键字，我们可以编写出能够同时处理成千上万个I/O操作的程序，而无需创建同等数量的线程或进程。本篇博客将深入探讨Python的异步编程模型，并结合FastAPI、Sanic、Quart等主流异步Web框架，通过详细的案例和解析，带您领略异步世界的强大威力。

---

第一部分：异步编程核心：asyncio与async/await

要理解异步框架，首先必须掌握其底层基石——asyncio。

1.1 概念解析

• 协程 (Coroutine): 一种特殊的函数，它可以在执行过程中暂停，并在稍后恢复。这是异步编程的核心。定义一个协程函数，需要在def前加上async关键字。
• 事件循环 (Event Loop): 异步程序的“发动机”。它负责管理和调度所有的协程，当一个协程遇到await（表示一个可能耗时的I/O操作）时，事件循环会暂停它，并立即切换到执行其他就绪的协程，从而实现了单线程内的并发。
• await: 关键字，只能在async函数内部使用。它用于等待一个协程或一个可等待对象（Awaitable）的完成，其作用类似于同步代码中的=，但它不会阻塞整个线程。

1.2 实战案例：理解异步与同步的差异

让我们通过一个简单的例子来对比同步和异步代码的性能差异。

同步版本 (同步请求多个URL)

import time
import requests

def fetch_sync(url):
    """同步地获取一个URL的内容"""
    print(f"Starting sync fetch for {url}")
    response = requests.get(url)
    print(f"Finished sync fetch for {url}, status: {response.status_code}")
    return response.status_code

def main_sync():
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/1',
    ]
    start_time = time.time()
    for url in urls:
        fetch_sync(url)
    end_time = time.time()
    print(f"Sync version took: {end_time - start_time:.2f} seconds")

# main_sync()
# 输出大约为 3秒，因为请求是串行执行的

异步版本 (异步并发请求多个URL)

import asyncio
import aiohttp

async def fetch_async(session, url):
    """异步地获取一个URL的内容"""
    print(f"Starting async fetch for {url}")
    async with session.get(url) as response:
        print(f"Finished async fetch for {url}, status: {response.status}")
        return response.status

async def main_async():
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/1',
    ]
    # aiohttp.ClientSession 是异步HTTP客户端
    async with aiohttp.ClientSession() as session:
        start_time = time.time()
        # asyncio.gather 将多个协程打包，并发执行
        tasks = [fetch_async(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        end_time = time.time()
        print(f"Async version took: {end_time - start_time:.2f} seconds")
        return results

# 运行异步主函数
# asyncio.run(main_async())
# 输出大约为 1秒，因为三个请求是并发执行的

案例解析:

• 同步: fetch_sync函数执行时，必须等待requests.get完成才能返回，程序在此期间无法做任何事。处理3个请求需要3秒。
• 异步: fetch_async函数遇到await session.get(url)时，它会向事件循环注册一个“当我完成时请通知我”的任务，然后立即返回，让出控制权。事件循环接着执行下一个任务。当所有请求都完成后，gather会收集结果。处理3个请求仅需约1秒。
• 核心: 异步编程的精髓在于将等待I/O的时间利用起来，去做其他事情，从而大幅提高I/O密集型任务的吞吐量。

---

第二部分：异步Web框架——构建高性能API

当我们将异步编程的思想应用于Web开发，就诞生了异步Web框架。它们能够在单个进程中处理成千上万的并发连接，这对于构建API、微服务和实时应用至关重要。

2.1 FastAPI - 现代、高性能的异步框架

FastAPI (pip install fastapi[all]) 是目前最受欢迎的异步Web框架之一。它不仅性能卓越，还因为其强大的类型提示集成和自动生成文档的能力而备受推崇。

实战案例：使用FastAPI构建一个异步数据聚合API

假设我们需要一个API，它需要从外部API获取天气数据，再从数据库获取用户信息，最后将两者合并返回。这是一个典型的I/O密集型任务。

from fastapi import FastAPI
import asyncio
import httpx # 一个异步HTTP客户端
import time

app = FastAPI(title="Async Data Aggregator API")

# 模拟一个异步数据库查询
async def get_user_data_from_db(user_id: int):
    # 模拟数据库延迟
    await asyncio.sleep(0.5) 
    return {"id": user_id, "name": "Alice", "location": "New York"}

# 模拟一个异步外部API调用
async def get_weather_data(location: str):
    # 使用httpx.AsyncClient进行异步请求
    async with httpx.AsyncClient() as client:
        # 这里使用httpbin来模拟一个延迟的API
        resp = await client.get(f'https://httpbin.org/delay/1?loc={location}')
        # 模拟解析API返回的天气数据
        return {"location": location, "temp_celsius": 22, "status": "Sunny"}

@app.get("/user-weather/{user_id}")
async def get_user_weather(user_id: int):
    start_time = time.time()
    
    # 并发地执行两个I/O操作
    user_task = asyncio.create_task(get_user_data_from_db(user_id))
    weather_task = asyncio.create_task(get_weather_data("Beijing")) # 示例地点
    
    # 等待两个任务完成
    user_data = await user_task
    weather_data = await weather_task
    
    end_time = time.time()
    
    return {
        "user": user_data,
        "weather": weather_data,
        "processing_time": round(end_time - start_time, 2)
    }

# 运行: uvicorn main:app --reload

代码解析:

• async def get_user_weather(...): FastAPI的路由函数可以是异步的，这使得整个请求处理链路都可以是异步的。
• asyncio.create_task(): 将协程包装成任务（Task），并立即调度执行。这使得get_user_data_from_db和get_weather_data可以并发运行。
• await user_task 和 await weather_task: 等待两个并发任务的结果。
• 结果: 如果这两个I/O操作分别耗时0.5秒和1秒，同步执行需要1.5秒，而通过create_task并发执行，总耗时仅为1秒左右，性能提升显著。

2.2 Quart - Flask的异步兄弟

Quart (pip install quart) 是一个与Flask API完全兼容，但基于asyncio构建的框架。如果你想将一个现有的Flask应用迁移到异步模式，或者喜欢Flask的简洁风格但又需要异步性能，Quart是完美的选择。

实战案例：使用Quart重写一个Flask风格的异步API

from quart import Quart, jsonify
import asyncio
import aiohttp

app = Quart(__name__)

# 模拟一个耗时的异步任务
async def slow_task(task_id: int):
    await asyncio.sleep(1)
    return f"Task {task_id} completed!"

@app.route('/run-task/<int:task_id>')
async def run_task(task_id):
    result = await slow_task(task_id)
    return jsonify({"result": result})

@app.route('/run-many-tasks')
async def run_many_tasks():
    # 并发运行多个任务
    tasks = [slow_task(i) for i in range(5)]
    results = await asyncio.gather(*tasks)
    return jsonify({"results": results})

# 运行: hypercorn main:app

代码解析:

• from quart import Quart, jsonify: 导入方式几乎与Flask相同。
• async def run_task(...): 路由函数同样是异步的。
• hypercorn: Quart推荐的ASGI服务器，用于运行异步应用。

2.3 Sanic - 高性能的异步先行者

Sanic (pip install sanic) 是较早出现的Python异步Web框架，以其追求极致性能的目标而闻名。虽然FastAPI在功能和生态上后来居上，但Sanic在某些特定的性能测试中依然表现出色。

实战案例：使用Sanic创建一个实时计数器API

from sanic import Sanic
from sanic.response import json
import asyncio

app = Sanic("counter_app")
# 使用一个简单的内存变量作为计数器
counter = 0

@app.route("/count")
async def get_count(request):
    global counter
    # 模拟一个异步操作，比如更新数据库
    await asyncio.sleep(0.1) 
    counter += 1
    return json({"current_count": counter})

@app.websocket('/ws')
async def feed(request, ws):
    while True:
        data = f"Current count: {counter}"
        await ws.send(data)
        await asyncio.sleep(1) # 每秒推送一次

# 运行: sanic main:app --dev

代码解析:

• @app.websocket: Sanic对WebSocket的支持非常直接，非常适合构建实时应用。
• sanic: 启动Sanic应用的命令。

---

第三部分：异步编程的陷阱与最佳实践

虽然异步编程强大，但也容易踩坑。

常见陷阱:

1. 混合同步与异步: 在异步函数中调用同步的阻塞操作（如time.sleep或requests.get）会阻塞整个事件循环。必须使用对应的异步版本（如asyncio.sleep或aiohttp.Client）。
2. 忘记await: 如果你调用了一个协程但忘记了await，你实际上得到的是一个协程对象，而不是它的执行结果。
3. 过度使用: 对于CPU密集型任务，异步并不能带来好处，反而可能因为事件循环的调度开销而变慢。此时应考虑使用多进程（multiprocessing）。

最佳实践:

1. 拥抱类型提示: 在FastAPI等框架中，类型提示不仅能提供运行时验证，还能自动生成文档。
2. 使用异步库: 与数据库交互时，使用aiopg (PostgreSQL), aiomysql (MySQL)等异步驱动。
3. 理解生命周期: 学会使用lifespan管理应用的启动和关闭逻辑，例如初始化数据库连接池。