Python异步编程：用asyncio重构Web API，并发性能提升4.7倍

请求 → Sanic (asyncio event loop)
         ├─ asyncmy连接池 → MySQL
         ├─ aiohttp Session → 下游Java服务
         └─ asyncio.to_thread → 线程池处理CPU密集段

# app_sync.py - Flask 同步版本
from flask import Flask, jsonify, request
import pymysql
import requests
import time

app = Flask(__name__)

# 全局MySQL连接（实际上Gunicorn每个worker一个独立连接）
db = pymysql.connect(host='10.0.0.1', user='app', password='xxx',
                     database='orders', charset='utf8mb4')

@app.route('/api/v1/orders', methods=['GET'])
def get_orders():
    user_id = request.args.get('user_id')
    start = time.time()

    try:
        with db.cursor() as cur:
            # 同步阻塞查询，耗时约80ms
            cur.execute("SELECT order_id, amount, status FROM orders WHERE user_id=%s ORDER BY create_time DESC LIMIT 20", (user_id))
            orders = cur.fetchall()

        # 同步调用下游服务，耗时平均150ms
        resp = requests.get(f"http://user-service:8080/api/users/{user_id}", timeout=1.5)
        user_info = resp.json()

        return jsonify({"code": 0, "data": {"orders": orders, "user": user_info}})

    except Exception as e:
        return jsonify({"code": 500, "msg": str(e)}), 500

# 运行：gunicorn -w 4 --threads 8 -b 0.0.0.0:5000 app_sync:app

# app_async.py - Sanic 异步版本
from sanic import Sanic, json
from sanic.request import Request
import asyncmy
import aiohttp
import asyncio
import uvloop
import re

app = Sanic("order_api")

# 全局连接池 - asyncmy 支持真正的异步连接池
db_pool = None
# 全局复用 aiohttp Session，避免TCP三次握手开销
http_session = None

@app.listener('before_server_start')
async def init_db(app, loop):
    global db_pool, http_session
    # uvloop 替换默认事件循环，Linux 下性能提升 20%-30%
    uvloop.install()
    # 连接池：最小5，最大20，注意 maxsize 要评估并发量
    db_pool = await asyncmy.create_pool(
        host='10.0.0.1', user='app', password='xxx',
        database='orders', charset='utf8mb4',
        minsize=5, maxsize=20, autocommit=True,
        pool_recycle=3600  # 连接存活1小时后自动回收
    )
    # aiohttp 连接池：每主机最大100连接
    conn = aiohttp.TCPConnector(limit=100, ttl_dns_cache=300)
    http_session = aiohttp.ClientSession(connector=conn, timeout=aiohttp.ClientTimeout(total=1.5))

@app.post('/api/v1/orders')
async def get_orders(request: Request):
    user_id = request.args.get('user_id')

    # 1. 异步MySQL查询 - 事件循环不会阻塞
    async with db_pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute(
                "SELECT order_id, amount, status FROM orders WHERE user_id=%s ORDER BY create_time DESC LIMIT 20",
                (user_id)
            )
            orders = await cur.fetchall()

    # 2. 异步HTTP调用下游服务
    async with http_session.get(f"http://user-service:8080/api/users/{user_id}") as resp:
        user_info = await resp.json()

    # 3. CPU密集部分（订单金额格式化+JSON序列化）丢线程池
    def process_data(orders, user_info):
        # 模拟正则清洗和金额格式化，约5ms
        for o in orders:
            o['amount'] = f"¥{o['amount']:.2f}"
        return {"orders": orders, "user": user_info}

    result = await asyncio.to_thread(process_data, orders, user_info)

    return json({"code": 0, "data": result})

# 运行：sanic app_async:app --host=0.0.0.0 --port=8000 --workers=4 --debug=False