python - 如何解决 Flask-SQLAlchemy 数据库 session 超时(避免 "MySQL server has gone away")

Question

我在这里多次看到这个问题，但答案对我没有帮助。

根据我的理解，Flask-SQLAlchemy 在 Flask 请求开始时连接到数据库，并在完成后拆除此 session 。但如果这个请求需要很长时间怎么办？

例如，我的应用程序接受一个 POST 请求，该请求启动一个需要写入数据库的长时间运行的 Celery 任务。 MySQL 服务器在一段时间后“消失”，使得任务结束时写回数据库不起作用。

我在网上阅读设置 options["pool_pre_ping"] = True 来缓解此类问题(使用发现的子类化 SQLAlchemy 解决方法 here )，但这没有什么区别。发生同样的错误，并且几乎似乎没有任何设置实际上发生改变。

我认为解决方案是重新考虑任务如何写入数据库。目前，我正在使用任务请求中的相同 session 。如果这是错误的，有什么想法吗？

Answer 1

我认为您对于重新思考任务如何写入数据库的想法是正确的。

让 Flask 路由将任务附加到队列中可能是谨慎的做法。一旦任务进入队列，Flask 路由就可以返回，从而关闭 Web 请求上下文。

然后，您可以在后台运行另一个 python(非 Flask)程序，该程序不断从此队列中读取数据，并从中弹出最新任务，然后执行任务并将它们保存到数据库中。

这可以使用以下方法完成:也许您已经在使用 Celery 调用异步函数来执行此操作？这是一篇关于如何创建此类 Celery 任务并将其与 Flask 路由一起使用的博文 https://blog.miguelgrinberg.com/post/using-celery-with-flask

你是对的，当 Web 请求上下文关闭时，Flask-SQLAlchemy 会破坏数据库 session ，这会在 Flask 路由返回时发生。

为了防止 Flask-SQLAlchemy 关闭 session ，您不应将 session 从 Flask 路由传递到异步 Celery 函数。相反，您应该在 celery 异步函数中“创建一个新 session ”(调用 session 注册表)。

事实上，这是在 Web 应用程序中使用 SQLAlchemy session 时的一般原则。您希望避免传递 session 对象。 SQLAlchemy 通过使用注册表模式创建返回“线程本地” session 对象的 session 注册表来推广此原则。 “线程本地” session 的想法意味着，当您从 session 注册表中获取 session 时，您将确保该 session 仅在当前线程中使用。由于您最终要创建一个新线程来执行长时间运行的任务，因此您可以依靠 session 注册表为您提供一个不会受到 Web 请求关闭影响的线程，因为 Web 请求是在不同的线程中处理的.

Flask-SQLAlchemy 默认实现此 session 注册表。例如，您可能有如下所示的路线和 celery 任务:

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from celery import Celery

app = Flask(__name__)
db = SQLAlchemy(app)

app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)


@celery.task
def do_stuff_that_takes_10_minutes(payload):
    session = db.session  # Gets a thread local session from the Session Registry that is different from the session obj you used in the route because this function will be executed in its own thread.
    # some long running task here
    return result

@app.route('/long_running_task', methods=['GET'])
def long_running_task():
  session = db.session  # Get a thread local session from the Session Registry that will close once this web request is complete

  # use this session here and do stuff

  # Call this function, which Celery will spawn a new thread to do
  # Don't pass the above session object into this function, we want this
  # session obj to stay only within this thread.
  do_stuff_that_takes_10_minutes.delay()

  return "Executing task, could take 10 minutes or more."