asyncio重构Flask接口：并发等待缩短3.2倍响应时间

改造前：MySQL查订单 → HTTP用户服务 → HTTP商品服务 → HTTP库存服务 → MySQL查扩展表
改造后：MySQL查订单 → 并发发起（HTTP用户服务 | HTTP商品服务 | HTTP库存服务 | MySQL查扩展表）

# 改造前：串行调用，总耗时 = sum(单个耗时)
import psycopg2
import requests

def get_order_detail(order_id):
    # 1. 查MySQL订单主记录
    conn = psycopg2.connect("dbname=orders user=postgres")
    cur = conn.cursor()
    cur.execute("SELECT user_id, product_id, stock_id FROM orders WHERE id=%s", (order_id,))
    order = cur.fetchone()

    # 2. 调用用户服务
    user_resp = requests.get(f"http://user-service/api/users/{order[0]}", timeout=5)
    user_data = user_resp.json()

    # 3. 调用商品服务
    product_resp = requests.get(f"http://product-service/api/products/{order[1]}", timeout=5)
    product_data = product_resp.json()

    # 4. 调用库存服务
    stock_resp = requests.get(f"http://stock-service/api/stocks/{order[2]}", timeout=5)
    stock_data = stock_resp.json()

    # 5. 再查一次MySQL扩展表
    cur.execute("SELECT remark FROM order_ext WHERE order_id=%s", (order_id,))
    remark = cur.fetchone()
    conn.close()

    return {"user": user_data, "product": product_data, 
            "stock": stock_data, "remark": remark}

# 改造后：并发发起所有独立请求
import asyncio
import asyncpg
import httpx

# 事件循环中创建全局连接池（避免每次请求重建连接）
async def init_pool():
    return await asyncpg.create_pool(
        "postgresql://postgres@localhost/orders",
        min_size=5, max_size=20,
        command_timeout=10  # 秒
    )

# 信号量限制并发数，防止打爆下游
_semaphore = asyncio.Semaphore(50)

async def fetch_with_limit(client, url):
    async with _semaphore:
        resp = await client.get(url, timeout=5.0)
        return resp.json()

async def get_order_detail_async(pool, order_id, client):
    # 1. 先查MySQL订单主记录（必须串行，因为后面需要它的值）
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT user_id, product_id, stock_id FROM orders WHERE id=$1", order_id
        )

    # 2. 并发发起三个HTTP调用和一个DB查询
    user_url = f"http://user-service/api/users/{row['user_id']}"
    product_url = f"http://product-service/api/products/{row['product_id']}"
    stock_url = f"http://stock-service/api/stocks/{row['stock_id']}"

    # 用TaskGroup管理协程，3.11+特性，异常处理更优雅
    async with asyncio.TaskGroup() as tg:
        task_user = tg.create_task(fetch_with_limit(client, user_url))
        task_product = tg.create_task(fetch_with_limit(client, product_url))
        task_stock = tg.create_task(fetch_with_limit(client, stock_url))
        task_remark = tg.create_task(
            pool.fetchval("SELECT remark FROM order_ext WHERE order_id=$1", order_id)
        )

    return {
        "user": task_user.result(),
        "product": task_product.result(),
        "stock": task_stock.result(),
        "remark": task_remark.result()
    }

# Flask视图函数入口
def get_order_detail_view(order_id):
    # 每次请求创建新的事件循环（Flask线程模型下最简单的方式）
    loop = asyncio.new_event_loop()
    try:
        asyncio.set_event_loop(loop)
        return loop.run_until_complete(
            get_order_detail_async(global_pool, order_id, global_client)
        )
    finally:
        loop.close()

import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())