asyncio重构Flask API：请求耗时从2.3秒降至480毫秒

# 原始同步实现（简化版）
def aggregate_report(user_id: str):
    profile = requests.get(f"http://profile-svc/user/{user_id}", timeout=3).json()
    orders = requests.get(f"http://order-svc/stats/{user_id}", timeout=3).json()
    risk = requests.get(f"http://risk-svc/tags/{user_id}", timeout=3).json()
    return {"profile": profile, "orders": orders, "risk": risk}

import asyncio
import aiohttp
from flask import Flask, jsonify

app = Flask(__name__)
# 全局session，复用连接池
_session = aiohttp.ClientSession(
    connector=aiohttp.TCPConnector(limit=100, ttl_dns_cache=300),
    timeout=aiohttp.ClientTimeout(total=2)
)

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

async def aggregate_async(user_id: str):
    tasks = [
        fetch_json(_session, f"http://profile-svc/user/{user_id}"),
        fetch_json(_session, f"http://order-svc/stats/{user_id}"),
        fetch_json(_session, f"http://risk-svc/tags/{user_id}"),
    ]
    results = await asyncio.gather(*tasks, return_exceptions=False)
    return {
        "profile": results[0],
        "orders": results[1],
        "risk": results[2]
    }

@app.route('/api/v1/report/aggregate')
def aggregate_report():
    user_id = request.args.get('user_id')
    try:
        result = asyncio.run(aggregate_async(user_id))
        return jsonify(result)
    except Exception as e:
        return jsonify({"error": str(e)}), 500

import asyncio
import aiohttp
import time
import logging
from flask import Flask, request, jsonify
from asgiref.sync import async_to_sync

logger = logging.getLogger(__name__)

# ------ 配置区 ------
DOWNSTREAM_TIMEOUT = 2.0  # 单次下游请求超时（秒）
MAX_CONCURRENCY = 50      # 全局并发信号量
CONNECTION_LIMIT = 100    # 连接池上限

# ------ 全局资源 ------
_semaphore = asyncio.Semaphore(MAX_CONCURRENCY)

_session = aiohttp.ClientSession(
    connector=aiohttp.TCPConnector(
        limit=CONNECTION_LIMIT,
        ttl_dns_cache=300,  # DNS缓存5分钟，减少DNS查询
        force_close=False,  # 保持keep-alive
    ),
    timeout=aiohttp.ClientTimeout(total=DOWNSTREAM_TIMEOUT)
)

async def fetch_with_semaphore(session, url):
    """带信号量限流的GET请求"""
    async with _semaphore:
        try:
            async with session.get(url) as resp:
                if resp.status == 200:
                    return await resp.json()
                else:
                    logger.warning(f"下游返回非200: {url} -> {resp.status}")
                    return None
        except asyncio.TimeoutError:
            logger.error(f"下游超时: {url}")
            return None
        except Exception as e:
            logger.error(f"下游请求异常: {url} -> {str(e)}")
            return None

async def aggregate_async(user_id: str):
    """并发聚合三个下游，返回dict或None"""
    urls = [
        f"http://profile-svc/user/{user_id}",
        f"http://order-svc/stats/{user_id}",
        f"http://risk-svc/tags/{user_id}",
    ]
    # 每个任务独立超时控制，避免一个慢任务拖死全部
    tasks = [asyncio.wait_for(fetch_with_semaphore(_session, url), timeout=DOWNSTREAM_TIMEOUT) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # 过滤掉异常和None，但保留部分结果
    valid_results = [r for r in results if isinstance(r, dict)]
    if not valid_results:
        raise RuntimeError("所有下游服务均不可用")

    return {
        "profile": results[0] if isinstance(results[0], dict) else {"error": "profile_unavailable"},
        "orders": results[1] if isinstance(results[1], dict) else {"error": "orders_unavailable"},
        "risk": results[2] if isinstance(results[2], dict) else {"error": "risk_unavailable"},
        "partial": len(valid_results) < 3
    }

# 使用async_to_sync桥接，保持Flask视图函数同步风格
@app.route('/api/v1/report/aggregate')
def aggregate_report():
    user_id = request.args.get('user_id')
    start = time.perf_counter()
    try:
        # 每次请求创建新的事件循环，避免跨请求状态污染
        result = asyncio.run(aggregate_async(user_id))
        elapsed = time.perf_counter() - start
        logger.info(f"聚合接口耗时: {elapsed*1000:.1f}ms, user_id={user_id}")
        return jsonify(result)
    except Exception as e:
        logger.error(f"聚合失败: {str(e)}")
        return jsonify({"error": "internal_error"}), 500