asyncio重构Flask接口：Web API并发量提升3倍实录

# app/routes.py (优化前)
import requests
from flask import Flask, jsonify

app = Flask(__name__)
SERVICE_A = "http://svc-a:8080/order"
SERVICE_B = "http://svc-b:8080/user"
SERVICE_C = "http://svc-c:8080/stock"

@app.route("/aggregate/")
def aggregate(order_id):
    # 串行调用，总耗时 ≈ 3 * 800ms = 2.4s
    order_data = requests.get(f"{SERVICE_A}/{order_id}", timeout=2).json()
    user_data = requests.get(f"{SERVICE_B}/{order_data['uid']}", timeout=2).json()
    stock_data = requests.get(f"{SERVICE_C}/{order_id}", timeout=2).json()
    return jsonify({"order": order_data, "user": user_data, "stock": stock_data})

Flask视图(同步) --> asyncio.run(fetch_all()) 
    --> fetch_all: gather(3个协程, 受semaphore控制)
        --> aiohttp.ClientSession.get(SERVICE_A/B/C) 
            --> 复用全局TCPConnector连接池

# app/services.py (优化后)
import asyncio
import aiohttp
from flask import Flask, jsonify, request
import time

app = Flask(__name__)

# 全局连接池：限制单host最大连接数100，开启keepalive
connector = aiohttp.TCPConnector(limit=100, limit_per_host=50, ttl_dns_cache=300)
session = aiohttp.ClientSession(connector=connector, timeout=aiohttp.ClientTimeout(total=2))

# 信号量：限制并发外部请求总数，防抖保护下游
_SEMAPHORE = asyncio.Semaphore(50)

SERVICE_A = "http://svc-a:8080/order"
SERVICE_B = "http://svc-b:8080/user"
SERVICE_C = "http://svc-c:8080/stock"

async def fetch_json(client_session, url):
    """协程函数：带信号量保护的异步GET请求"""
    async with _SEMAPHORE:
        async with client_session.get(url) as resp:
            if resp.status != 200:
                raise RuntimeError(f"HTTP {resp.status} for {url}")
            return await resp.json()

async def fetch_all(order_id, user_id):
    """核心：并发发起3个外部HTTP请求"""
    async with aiohttp.ClientSession() as fresh_session:  # 为了演示简洁，实际可复用全局session
        # 注意：这里使用全局session更优，但避免单点连接池共享问题
        results = await asyncio.gather(
            fetch_json(session, f"{SERVICE_A}/{order_id}"),
            fetch_json(session, f"{SERVICE_B}/{user_id}"),
            fetch_json(session, f"{SERVICE_C}/{order_id}"),
            return_exceptions=False  # 只要有一个失败，整体失败
        )
        return results

@app.route("/aggregate/")
def aggregate(order_id):
    # 从请求上下文获取user_id（实际业务中可能从token解析）
    user_id = request.headers.get("X-User-ID", "default")

    # 关键点：Flask同步视图内运行asyncio事件循环
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        order_data, user_data, stock_data = loop.run_until_complete(
            fetch_all(order_id, user_id)
        )
    finally:
        loop.close()

    return jsonify({"order": order_data, "user": user_data, "stock": stock_data})

@app.after_request
def close_session(response):
    # 注意：生产环境千万别在每次请求后close session！
    return response

# app/services.py (最终版本 - 去掉跨循环信号量，改用连接池限流)
import asyncio, aiohttp
from flask import Flask, jsonify

app = Flask(__name__)

# 每个host最大50并发，超过的请求自动排队
connector = aiohttp.TCPConnector(limit=100, limit_per_host=50, ttl_dns_cache=300)
session = aiohttp.ClientSession(connector=connector, timeout=aiohttp.ClientTimeout(total=2))

async def fetch_json(url):
    async with session.get(url) as resp:
        resp.raise_for_status()
        return await resp.json()

async def fetch_all(order_id, user_id):
    return await asyncio.gather(
        fetch_json(f"{SERVICE_A}/{order_id}"),
        fetch_json(f"{SERVICE_B}/{user_id}"),
        fetch_json(f"{SERVICE_C}/{order_id}"),
    )

@app.route("/aggregate/")
def aggregate(order_id):
    user_id = request.headers.get("X-User-ID", "default")
    order_data, user_data, stock_data = asyncio.run(fetch_all(order_id, user_id))
    return jsonify({"order": order_data, "user": user_data, "stock": stock_data})

import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# 这样asyncio.run()才会用uvloop