Flask多路由复用重计算函数的性能优化方案咨询

这个问题我之前也碰到过，重复跑重计算逻辑确实会让应用变慢不少，下面给你几个靠谱的解决思路，你可以根据自己的场景选：

1. 用Flask-Caching扩展（最推荐，适合生产环境） #

这是最成熟的方案，支持多种缓存后端，还能自动管理缓存失效，不用自己造轮子。

步骤： #

• 先安装扩展：

  pip install flask-caching
  

• 然后修改你的代码，初始化缓存并给重计算函数加上缓存装饰器：

from flask import Flask, render_template, request, session, url_for, Markup, send_from_directory
from flask_caching import Cache
from AnotherModule import perform_heavy_calculations, convert_dataframe

application = Flask(__name__)
application.secret_key = 'your_secure_secret_key_here'  # 确保session正常工作

# 配置缓存：开发环境用内存缓存，生产可以换成Redis/Memcached
cache_config = {
    "CACHE_TYPE": "SimpleCache",
    "CACHE_DEFAULT_TIMEOUT": 300  # 缓存5分钟，可根据业务调整
}
cache = Cache(application, config=cache_config)

# 给重计算函数套一层缓存，按keyword作为缓存键
@cache.memoize()
def cached_heavy_calc(keyword):
    # 调用原函数，不用修改原来的perform_heavy_calculations代码
    return perform_heavy_calculations(keyword)

@application.route('/<string:keyword>', methods=['GET'])
def index(keyword):
    dataframe = cached_heavy_calc(keyword)
    session['keyword'] = keyword
    return render_template("index.html", dataframe=dataframe)

@application.route("/timeseries.json", methods=['GET'])
def timeseries():
    # 注意：原代码里这个路由的参数写错了，路由规则里没有keyword，所以要去掉函数参数
    keyword = session.get('keyword')
    if not keyword:
        return "Keyword not found in session", 400
    dataframe = cached_heavy_calc(keyword)
    timeseries_data = convert_dataframe(dataframe)
    return render_template("timeseries.html", timeseries=timeseries_data)

if __name__ == '__main__':
    application.run(debug=True, threaded=True)

优点： #

• 不用改动原有的重计算函数，侵入性低
• 支持自动超时失效，避免缓存数据太旧
• 生产环境可以轻松切换到Redis/Memcached等分布式缓存，适配多进程/多服务器部署

2. 用Python内置的functools.lru_cache（轻量，适合小型应用/开发环境） #

如果你的应用比较小，或者只是开发环境用，这个方案不用装额外库，非常方便。

from flask import Flask, render_template, request, session, url_for, Markup, send_from_directory
from functools import lru_cache
from AnotherModule import perform_heavy_calculations, convert_dataframe

application = Flask(__name__)
application.secret_key = 'your_secure_secret_key_here'

# 给重计算函数加上LRU缓存，maxsize限制缓存条目数，防止内存溢出
@lru_cache(maxsize=128)
def cached_heavy_calc(keyword):
    return perform_heavy_calculations(keyword)

# 后面的路由代码和上面一样...

注意事项： #

• 这个缓存是进程内的，如果生产环境用Gunicorn多进程部署，每个进程会有自己的缓存副本，可能导致重复计算
• 没有自动超时失效，需要手动调用cached_heavy_calc.cache_clear()来清空缓存

3. 全局字典+线程锁（不推荐，除非你必须自己实现） #

这个方案需要自己管理缓存和线程安全，容易出问题，除非万不得已不建议用：

from flask import Flask, render_template, request, session, url_for, Markup, send_from_directory
from AnotherModule import perform_heavy_calculations, convert_dataframe
import threading

application = Flask(__name__)
application.secret_key = 'your_secure_secret_key_here'

# 全局缓存字典和线程锁，防止并发访问出问题
calc_cache = {}
cache_lock = threading.Lock()

def get_cached_dataframe(keyword):
    with cache_lock:
        if keyword not in calc_cache:
            calc_cache[keyword] = perform_heavy_calculations(keyword)
        return calc_cache[keyword]

# 路由里调用get_cached_dataframe(keyword)即可

缺点： #

• 需要自己处理缓存失效（比如定期清理旧数据）
• 线程锁可能会影响并发性能
• 多进程部署时缓存不共享

另外，原代码里有个小bug：timeseries路由的函数定义里写了keyword参数，但路由规则/timeseries.json里没有这个参数，会导致Flask报错，我在上面的代码里已经修正了这个问题。

内容的提问来源于stack exchange，提问作者Antonio