Flask构建高性能Web服务的技术方案，涵盖异步处理、数据库优化、缓存策略、任务队列

from flask import Flask
app = Flask(__name__)

@app.route('/api/data')
def get_data():
    # 模拟耗时数据库查询
    import time
    time.sleep(0.5)  
    return {'status': 'success'}

if __name__ == '__main__':
    app.run(threaded=True)  # 默认单进程多线程模式

# gunicorn.conf.py
workers = 4               # CPU核心数 × 2
worker_class = 'gevent'   # 使用协程模型
worker_connections = 2000 # 单个worker最大连接数
timeout = 30              # 超时时间（秒）
keepalive = 2             # 长连接保持时间

gunicorn -c gunicorn.conf.py app:app

from flask_sqlalchemy import SQLAlchemy

app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/db'
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
    'pool_size': 20,          # 连接池大小
    'max_overflow': 10,       # 临时扩容连接数
    'pool_recycle': 300,      # 连接回收时间（秒）
    'pool_pre_ping': True     # 自动检测失效连接
}

db = SQLAlchemy(app)

@app.route('/users/<int:user_id>')
def get_user(user_id):
    # 错误示例：未使用ORM优化
    # user = User.query.get(user_id)  
    
    # 优化后的查询方式
    user = db.session.query(
        User.id, 
        User.username,
        User.email
    ).filter_by(id=user_id).first()
    
    return {'user': user._asdict()}

from flask_caching import Cache

cache = Cache(config={
    'CACHE_TYPE': 'RedisCache',
    'CACHE_REDIS_URL': 'redis://localhost:6379/0',
    'CACHE_DEFAULT_TIMEOUT': 300
})
cache.init_app(app)

@app.route('/hot-news')
@cache.cached(query_string=True)  # 根据查询参数缓存
def get_hot_news():
    # 复杂查询逻辑
    news = News.query.order_by(News.clicks.desc()).limit(10).all()
    return {'news': [n.to_dict() for n in news]}

# 缓存失效示例
def update_news(news_id):
    news = News.query.get(news_id)
    # 更新数据库后清除缓存
    cache.delete_memoized(get_hot_news)

from celery import Celery

celery = Celery(
    'tasks',
    broker='amqp://guest@localhost//',
    backend='redis://localhost:6379/1'
)

@celery.task
def process_image(image_path):
    # 耗时图像处理任务
    import PIL.Image
    img = PIL.Image.open(image_path)
    # 处理逻辑...
    return result_url

@app.route('/upload', methods=['POST'])
def upload_image():
    image_file = request.files['image']
    temp_path = save_temp_file(image_file)
    # 异步执行任务
    task = process_image.delay(temp_path)  
    return {'task_id': task.id}