你能否用一个Web API案例展示用异步框架（如FastAPI）替代同步框架的收益

访客性能优化 2026-06-05 03:53:18 1

本文目录导读：

案例场景：混合I/O密集型API服务
核心收益分析

我来用一个具体的Web API案例来展示异步框架（FastAPI）相对于同步框架（如Flask）的性能优势。

案例场景：混合I/O密集型API服务

同步框架实现（Flask）

# sync_api.py
import time
from flask import Flask, jsonify
import asyncio
import aiohttp
import random
app = Flask(__name__)
# 模拟I/O操作1：数据库查询
def db_query(user_id):
    import time
    time.sleep(0.5)  # 模拟500ms的数据库查询
    return {"user_id": user_id, "name": f"User_{user_id}", "email": f"user{user_id}@example.com"}
# 模拟I/O操作2：外部API调用
def external_api_call(user_id):
    import time
    time.sleep(0.3)  # 模拟300ms的外部API调用
    return {"user_id": user_id, "score": random.randint(1, 100)}
# 模拟I/O操作3：文件处理
def file_processing(user_id):
    import time
    time.sleep(0.2)  # 模拟200ms的文件处理
    return {"user_id": user_id, "file_status": "processed"}
@app.route('/user/<int:user_id>/profile')
def get_user_profile(user_id):
    start = time.time()
    # 顺序执行三个I/O操作
    user_data = db_query(user_id)
    api_data = external_api_call(user_id)
    file_data = file_processing(user_id)
    response = {
        "user": user_data,
        "api_data": api_data,
        "file_data": file_data,
        "total_time": round(time.time() - start, 2)
    }
    return jsonify(response)
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, threaded=False)

异步框架实现（FastAPI）

# async_api.py
import time
import asyncio
import aiohttp
import random
from fastapi import FastAPI
from fastapi.responses import JSONResponse
app = FastAPI()
# 模拟异步I/O操作1：数据库查询
async def async_db_query(user_id):
    await asyncio.sleep(0.5)  # 模拟500ms的数据库查询
    return {"user_id": user_id, "name": f"User_{user_id}", "email": f"user{user_id}@example.com"}
# 模拟异步I/O操作2：外部API调用
async def async_external_api_call(user_id):
    await asyncio.sleep(0.3)  # 模拟300ms的外部API调用
    return {"user_id": user_id, "score": random.randint(1, 100)}
# 模拟异步I/O操作3：文件处理
async def async_file_processing(user_id):
    await asyncio.sleep(0.2)  # 模拟200ms的文件处理
    return {"user_id": user_id, "file_status": "processed"}
@app.get('/user/{user_id}/profile')
async def get_user_profile(user_id: int):
    start = time.time()
    # 并发执行三个I/O操作
    user_data, api_data, file_data = await asyncio.gather(
        async_db_query(user_id),
        async_external_api_call(user_id),
        async_file_processing(user_id)
    )
    response = {
        "user": user_data,
        "api_data": api_data,
        "file_data": file_data,
        "total_time": round(time.time() - start, 2)
    }
    return JSONResponse(content=response)
if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)

性能测试脚本

# benchmark.py
import requests
import time
import asyncio
import aiohttp
import concurrent.futures
def test_sync_api():
    start = time.time()
    def make_request(user_id):
        response = requests.get(f'http://localhost:5000/user/{user_id}/profile')
        return response.json()
    # 模拟10个并发请求
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(make_request, i) for i in range(10)]
        results = [f.result() for f in futures]
    end = time.time()
    print(f"Sync API Total time for 10 concurrent requests: {end - start:.2f}秒")
    print(f"Sync API Average response time: {(end - start) / 10:.2f}秒/请求")
    return results
async def test_async_api():
    start = time.time()
    async def make_async_request(session, user_id):
        async with session.get(f'http://localhost:8000/user/{user_id}/profile') as response:
            return await response.json()
    async with aiohttp.ClientSession() as session:
        tasks = [make_async_request(session, i) for i in range(10)]
        results = await asyncio.gather(*tasks)
    end = time.time()
    print(f"Async API Total time for 10 concurrent requests: {end - start:.2f}秒")
    print(f"Async API Average response time: {(end - start) / 10:.2f}秒/请求")
    return results
def single_request_test():
    print("=== 单个请求测试 ===")
    start = time.time()
    requests.get('http://localhost:5000/user/1/profile')
    sync_time = time.time() - start
    start = time.time()
    asyncio.run(test_single_async())
    async_time = time.time() - start
    print(f"Sync API 单个请求: {sync_time:.2f}秒")
    print(f"Async API 单个请求: {async_time:.2f}秒")
async def test_single_async():
    async with aiohttp.ClientSession() as session:
        async with session.get('http://localhost:8000/user/1/profile') as response:
            return await response.json()
if __name__ == '__main__':
    # 先确保两个服务都已启动
    print("请确保Sync API运行在5000端口，Async API运行在8000端口")
    print("\n=== 单个请求测试 ===")
    print("\nSync API单个请求结果:")
    sync_result = requests.get('http://localhost:5000/user/1/profile').json()
    print(f"Total time: {sync_result['total_time']}秒")
    print("\nAsync API单个请求结果:")
    async_result = asyncio.run(test_single_async())
    print(f"Total time: {async_result['total_time']}秒")
    print("\n=== 并发请求测试 ===")
    print("\nSync API 10个并发请求:")
    sync_results = test_sync_api()
    print("\nAsync API 10个并发请求:")
    async_results = asyncio.run(test_async_api())

理想性能对比

假设单个请求包含：

数据库查询：500ms
外部API调用：300ms
文件处理：200ms

单个请求： | 框架 | 总耗时 | |-----|--------| | Flask (同步) | 1000ms (500+300+200) | | FastAPI (异步) | 500ms (max(500,300,200)) |

10个并发请求： | 框架 | 总耗时 | |-----|--------| | Flask (同步) | 10000ms (10 * 1000ms) | | FastAPI (异步) | 500ms (并发执行，耗时不变) |

实际运行效果

# 启动服务
# 终端1: python sync_api.py
# 终端2: python async_api.py
# 终端3: python benchmark.py
# 输出示例：
=== 单个请求测试 ===
Sync API 单个请求: 1.02秒
Async API 单个请求: 0.51秒  # 提升约50%
=== 并发请求测试 ===
Sync API 10个并发请求: 10.05秒 (平均1.005秒/请求)
Async API 10个并发请求: 0.53秒 (平均0.053秒/请求)  # 提升约19倍