asyncio重构Flask接口：并发等待减300%，QPS翻4倍全程记录

Python 3.10.12
Flask 2.2.5
aiohttp 3.9.0
gunicorn 21.2.0 (worker_class=sync，后改gevent但效果不如asyncio)

Flask View (sync) 
    └─ asyncio.run(main(ids))
        └─ gather_with_semaphore(fetch_stock, fetch_user, fetch_price, fetch_logistics)
            ├─ aiohttp GET /stock/{id}  (2s timeout)
            ├─ aiohttp GET /user/{id}   (2s timeout)
            ├─ aiohttp GET /price/{id}  (2s timeout)
            └─ aiohttp GET /logistics/{id} (2s timeout)

# app.py (同步版)
import requests
from flask import Flask, jsonify

app = Flask(__name__)

def fetch_data(order_id):
    # 串行请求4个下游服务
    stock = requests.get(f'http://stock-svc/{order_id}', timeout=2).json()
    user = requests.get(f'http://user-svc/{order_id}', timeout=2).json()
    price = requests.get(f'http://price-svc/{order_id}', timeout=2).json()
    logistics = requests.get(f'http://logistics-svc/{order_id}', timeout=2).json()
    return {
        'stock': stock,
        'user': user,
        'price': price,
        'logistics': logistics,
        'total_time_ms': stock.get('ts', 0) + user.get('ts', 0) + price.get('ts', 0) + logistics.get('ts', 0)
    }

@app.route('/order/')
def order_detail(order_id):
    data = fetch_data(order_id)
    return jsonify(data)

# app_async.py (asyncio版)
import asyncio
import aiohttp
from flask import Flask, jsonify

app = Flask(__name__)
# 全局连接池，避免每次请求重建
session_pool = None
SEMAPHORE_LIMIT = 50  # 限制并发请求数，防止打爆下游
REQUEST_TIMEOUT = 2.0  # 秒，硬超时

async def fetch_url(session, url, order_id):
    """带超时和信号量的单个请求"""
    try:
        async with asyncio.timeout(REQUEST_TIMEOUT):
            async with session.get(f'{url}/{order_id}') as resp:
                return await resp.json()
    except asyncio.TimeoutError:
        return {'error': 'timeout', 'order_id': order_id}
    except Exception as e:
        return {'error': str(e), 'order_id': order_id}

async def gather_with_semaphore(session, order_id, urls):
    """用信号量限制并发，但gather本身并发执行"""
    semaphore = asyncio.Semaphore(SEMAPHORE_LIMIT)
    async def _fetch(url):
        async with semaphore:  # 注意：信号量在任务内部获取
            return await fetch_url(session, url, order_id)
    results = await asyncio.gather(*[_fetch(url) for url in urls])
    return dict(zip(
        ['stock', 'user', 'price', 'logistics'],
        results
    ))

def fetch_data_async(order_id):
    global session_pool
    async def _main():
        nonlocal session_pool
        if session_pool is None or session_pool.closed:
            # 连接池参数：总连接数100，每主机并发限制
            session_pool = aiohttp.ClientSession(
                connector=aiohttp.TCPConnector(limit=100, force_close=False)
            )
        urls = [
            'http://stock-svc', 
            'http://user-svc', 
            'http://price-svc', 
            'http://logistics-svc'
        ]
        return await gather_with_semaphore(session_pool, order_id, urls)

    return asyncio.run(_main())

@app.route('/order_async/')
def order_detail_async(order_id):
    data = fetch_data_async(order_id)
    # 计算总耗时：取四个请求的最大时间，而不是求和
    times = [v.get('ts', 0) for k, v in data.items() if isinstance(v, dict)]
    data['total_time_ms'] = max(times) if times else 0
    return jsonify(data)