用asyncio重构Flask API：并发吞吐提升3倍，延迟降低60%

Python: 3.10.12 (CPython)
Web框架: Flask 2.3.3
WSGI服务器: Gunicorn 20.1.0
HTTP客户端: requests 2.31.0 (同步) / aiohttp 3.8.6 (异步)
操作系统: Ubuntu 20.04 LTS (Linux 5.4.0-144-generic)
压测工具: wrk 4.2.0

Flask视图函数
    │
    ▼
run_coroutine_threadsafe(async_main())
    │
    ▼
后台线程的事件循环
    ├── Task 1: 调用用户服务 (aiohttp)
    ├── Task 2: 调用订单服务 (aiohttp)
    └── Task 3: 调用库存服务 (aiohttp)

# app_sync.py
import requests
from flask import Flask, jsonify
import time

app = Flask(__name__)

def fetch_user():
    resp = requests.get('http://user-service:8080/api/user', timeout=3)
    return resp.json()

def fetch_orders():
    resp = requests.get('http://order-service:8080/api/orders', timeout=3)
    return resp.json()

def fetch_inventory():
    resp = requests.get('http://inventory-service:8080/api/inventory', timeout=3)
    return resp.json()

@app.route('/api/v1/user-summary')
def user_summary():
    start = time.perf_counter()
    user = fetch_user()
    orders = fetch_orders()
    inventory = fetch_inventory()
    result = {
        'user': user,
        'orders': orders,
        'inventory': inventory,
        'total_time_ms': int((time.perf_counter() - start) * 1000)
    }
    return jsonify(result)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

# app_async.py
import asyncio
import aiohttp
import time
from flask import Flask, jsonify

app = Flask(__name__)

# 全局配置
CONCURRENCY_LIMIT = 50  # 最大并发连接数
TIMEOUT = aiohttp.ClientTimeout(total=5, connect=2)  # 总超时5秒，连接超时2秒

# 全局事件循环和会话
loop = None
session = None
semaphore = None

def init_async_engine():
    """在后台线程中启动事件循环，并初始化全局ClientSession"""
    global loop, session, semaphore
    loop = asyncio.new_event_loop()
    session = aiohttp.ClientSession(loop=loop, timeout=TIMEOUT, connector=aiohttp.TCPConnector(limit=CONCURRENCY_LIMIT))
    semaphore = asyncio.Semaphore(CONCURRENCY_LIMIT)
    def run_loop():
        asyncio.set_event_loop(loop)
        loop.run_forever()
    import threading
    t = threading.Thread(target=run_loop, daemon=True)
    t.start()

async def fetch_url(session, url):
    """带信号量和超时控制的异步请求"""
    async with semaphore:
        try:
            async with session.get(url) as resp:
                if resp.status != 200:
                    return {'error': f'HTTP {resp.status}'}
                return await resp.json()
        except asyncio.TimeoutError:
            return {'error': 'timeout'}
        except aiohttp.ClientError as e:
            return {'error': str(e)}

async def fetch_all():
    """并发调用三个服务"""
    urls = [
        'http://user-service:8080/api/user',
        'http://order-service:8080/api/orders',
        'http://inventory-service:8080/api/inventory'
    ]
    results = await asyncio.gather(
        *[fetch_url(session, url) for url in urls],
        return_exceptions=True  # 防止一个失败导致全部失败
    )
    return results

@app.route('/api/v1/user-summary')
def user_summary():
    start = time.perf_counter()
    # 把协程提交到后台线程的事件循环，并等待结果
    future = asyncio.run_coroutine_threadsafe(fetch_all(), loop)
    user, orders, inventory = future.result(timeout=6)  # 最外层兜底超时
    result = {
        'user': user,
        'orders': orders,
        'inventory': inventory,
        'total_time_ms': int((time.perf_counter() - start) * 1000)
    }
    return jsonify(result)

if __name__ == '__main__':
    init_async_engine()
    app.run(host='0.0.0.0', port=5000, threaded=True)