asyncio重构Flask接口：请求耗时从800ms降至150ms

Python 3.8.10
Flask 2.0.3
aiohttp 3.8.1
asyncpg 0.25.0
redis-py 4.3.1 (使用其asyncio模式)

# 改造前：同步阻塞版本
from flask import Flask, jsonify
import requests
import pymysql
import redis

app = Flask(__name__)
r = redis.Redis(host='localhost', port=6379)

def get_order(order_id):
    # 1. 查缓存
    cached = r.get(f'order:{order_id}')
    if cached:
        return jsonify(eval(cached))
    # 2. 查数据库
    conn = pymysql.connect(host='localhost', user='root', password='123456', database='orders')
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM orders WHERE id=%s", (order_id,))
    order = cursor.fetchone()
    # 3. 调外部服务
    resp = requests.get(f'http://product-service/api/v1/product/{order[2]}', timeout=2)
    product = resp.json()
    return jsonify({'order': order, 'product': product})

@app.route('/api/v1/order/detail/')
def order_detail(order_id):
    return get_order(order_id)

# 改造后：asyncio异步版本
import asyncio
import aiohttp
import asyncpg
from redis.asyncio import Redis
from flask import Flask, jsonify

app = Flask(__name__)
redis_pool: Redis = None
pg_pool: asyncpg.Pool = None

async def init_pools():
    """初始化连接池，在应用启动时调用一次"""
    global redis_pool, pg_pool
    redis_pool = Redis(host='localhost', port=6379, decode_responses=True)
    pg_pool = await asyncpg.create_pool(
        host='localhost', 
        user='root', 
        password='123456',
        database='orders',
        min_size=10,      # 重要：最小连接数
        max_size=30,      # 重要：最大连接数，比原来80个线程的消耗低得多
        command_timeout=5 # 防止慢SQL拖死整个协程
    )

async def fetch_order_detail(order_id: int):
    # 1. 缓存查询
    cached = await redis_pool.get(f'order:{order_id}')
    if cached:
        return eval(cached)  # 生产环境建议用JSON模块

    # 2. 数据库查询 + 外部API调用，并发执行
    async with pg_pool.acquire() as conn:
        order_row = await conn.fetchrow(
            "SELECT * FROM orders WHERE id = $1", order_id
        )
        if not order_row:
            return None

    # 3. 并发调用商品服务，设置超时
    timeout = aiohttp.ClientTimeout(total=2)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(f'http://product-service/api/v1/product/{order_row["product_id"]}') as resp:
            product = await resp.json()

    result = {'order': dict(order_row), 'product': product}
    # 回填缓存，设置过期时间
    await redis_pool.set(f'order:{order_id}', repr(result), ex=60)
    return result

@app.route('/api/v1/order/detail/')
def order_detail(order_id):
    # Flask同步视图内调用事件循环
    loop = asyncio.new_event_loop()
    try:
        result = loop.run_until_complete(fetch_order_detail(order_id))
    finally:
        loop.close()
    return jsonify(result)

# 应用启动时初始化连接池
if __name__ == '__main__':
    asyncio.run(init_pools())  # 启动时创建连接池
    app.run(host='0.0.0.0', port=5000, threaded=False)  # 关键：关闭多线程模式

async def init_pools():
    global pg_pool, redis_pool
    pg_pool = await asyncpg.create_pool(...)
    # 预热：获取并归还连接，触发实际TCP连接建立
    for _ in range(10):
        async with pg_pool.acquire() as conn:
            await conn.execute('SELECT 1')  # 简单心跳

async def fetch_with_retry(session, url, retries=2):
    for i in range(retries):
        try:
            async with session.get(url) as resp:
                return await resp.json()
        except asyncio.TimeoutError:
            if i == retries - 1:
                raise
            await asyncio.sleep(0.1 * (i + 1))  # 退避重试