Python异步编程：用asyncio重构Flask接口，QPS翻3倍

同步版本：请求 → 线程池 → MySQL(100ms) → Redis(50ms) → HTTP(300ms) → 返回（总耗时450ms）
异步版本：请求 → 协程 → gather(MySQL(100ms), Redis(50ms), HTTP(300ms)) → 返回（总耗时300ms）

# profile_api_sync.py
import time
import requests
import redis

r = redis.Redis(host='redis-cache', port=6379, decode_responses=True)

def get_profile(user_id: str) -> dict:
    # 1. 查MySQL（用了ORM，这里简化为耗时模拟）
    time.sleep(0.1)  # 模拟DB查询 100ms
    user_info = {"user_id": user_id, "name": "张三"}

    # 2. 查Redis缓存
    tags = r.get(f"user:tags:{user_id}")  # 实际是hgetall，模拟耗时50ms

    # 3. 调风控HTTP接口
    risk_resp = requests.post(
        "http://risk-service/api/risk",
        json={"user_id": user_id},
        timeout=1.0
    ).json()  # 实际耗时300ms

    return {"user_info": user_info, "tags": tags, "risk_score": risk_resp["score"]}

# Flask路由
from flask import Flask, jsonify
app = Flask(__name__)

@app.route("/api/profile/")
def profile(user_id):
    return jsonify(get_profile(user_id))

# 压测：wrk -t4 -c100 -d30s http://service/api/profile/123
# 结果：QPS 210, P99 1.2s

# profile_api_async.py
import asyncio
import httpx
import redis.asyncio as aioredis
from asgiref.sync import async_to_sync

# 初始化异步Redis连接池
r = aioredis.from_url("redis://redis-cache:6379", decode_responses=True, max_connections=20)

# 全局httpx客户端，复用连接池（关键！）
client = httpx.AsyncClient(base_url="http://risk-service", timeout=httpx.Timeout(1.0))

# 信号量限制并发数，避免打爆下游服务
semaphore = asyncio.Semaphore(50)

async def fetch_mysql(user_id: int) -> dict:
    # 这里模拟DB查询，实际用aiomysql或asyncpg
    await asyncio.sleep(0.1)  # 实际场景：asyncpg执行SQL
    return {"user_id": user_id, "name": "张三"}

async def fetch_redis(user_id: int) -> str:
    # 异步Redis操作
    tags = await r.get(f"user:tags:{user_id}")
    return tags or "[]"

async def fetch_risk(user_id: int) -> float:
    # 异步HTTP调用，受信号量限制
    async with semaphore:
        resp = await client.post("/api/risk", json={"user_id": user_id})
        return resp.json()["score"]

async def get_profile_async(user_id: int) -> dict:
    # 并发执行三个IO
    user_info, tags, risk_score = await asyncio.gather(
        fetch_mysql(user_id),
        fetch_redis(user_id),
        fetch_risk(user_id)
    )
    return {"user_info": user_info, "tags": tags, "risk_score": risk_score}

# Flask路由（通过async_to_sync桥接）
from flask import Flask, jsonify
app = Flask(__name__)

@app.route("/api/profile/")
def profile(user_id):
    # 同步调用异步函数
    result = async_to_sync(get_profile_async)(user_id)
    return jsonify(result)

# 压测：wrk -t4 -c100 -d30s http://service/api/profile/123
# 结果：QPS 680, P99 320ms