使用tornado_sqlalchemy遇异常后会话无法恢复问题求助

问题背景 #

使用tornado_sqlalchemy管理数据库会话，替代原生SQLAlchemy解决了诸多问题，但在后台执行重查询（通过@run_on_executor装饰器）时，高负载场景下出现大量500错误，服务器最终挂死，必须重启才能恢复。

当前应用配置：

app = tornado.web.Application(
         ...
         db=SQLAlchemy(url=get_db_url(),
                       engine_options={
                         "pool_pre_ping": True,
                         "execution_options": {
                            "isolation_level": "AUTOCOMMIT"
                         }
                       }
    )

页面定义：

from tornado_sqlalchemy import SessionMixin
from concurrent.futures import ThreadPoolExecutor

class MyPage(tornado.web.RequestHandler, SessionMixin):
  # 后台执行任务
  executor = ThreadPoolExecutor(max_workers=10)

  def render_myhtml(self):
    # SessionMixin自动生成唯一会话self.session
    results = self.session.query(...)

报错信息 #

首个崩溃请求触发的连锁错误：
pymysql.err.InterfaceError: (0, '')

后续请求报错：
AttributeError: 'NoneType' object has no attribute 'read'

后续请求报错：
pymysql.err.InternalError: Packet sequence number wrong - got 102 expected 3

后续请求报错：
sqlalchemy.exc.PendingRollbackError: Can't reconnect until invalid transaction is rolled back.

最终触发连接池耗尽：
sqlalchemy.exc.TimeoutError: QueuePool limit of size 5 overflow 10 reached, connection timed out, timeout 30.00

复现步骤 #

在engine_options中添加以下配置：

pool_size = 2
max_overflow = 0

同时打开3个标签页即可触发问题。

核心原因：异常发生后，会话未被正确回滚、关闭，导致连接池资源被无效会话占用，最终耗尽所有连接，后续请求无法获取连接而挂死。

解决方案 #

1. 后台线程使用独立会话，添加异常处理 #

后台线程不能复用RequestHandler的self.session，必须单独创建会话，并在异常时强制回滚、关闭：

from tornado.concurrent import run_on_executor
from tornado_sqlalchemy import Session

class MyPage(tornado.web.RequestHandler, SessionMixin):
  executor = ThreadPoolExecutor(max_workers=10)

  @run_on_executor
  def background_query(self):
    session = None
    try:
      # 为后台线程创建独立会话
      session = Session(self.application.db.engine)
      results = session.query(...).all()
      return results
    except Exception:
      if session:
        session.rollback()
      raise
    finally:
      if session:
        session.close()

  async def get(self):
    results = await self.background_query()
    self.render("myhtml.html", results=results)

2. 优化连接池配置 #

在engine_options中添加连接回收和超时配置，避免无效连接占用资源：

engine_options={
  "pool_pre_ping": True,
  "execution_options": {
    "isolation_level": "AUTOCOMMIT"
  },
  "pool_recycle": 300,  # 每5分钟回收一次连接，避免数据库主动断开
  "pool_timeout": 10,   # 获取连接超时时间，防止请求无限等待
  "pool_size": 5,
  "max_overflow": 10
}

3. 自定义基类，强制清理会话 #

重写RequestHandler的异常处理和请求结束方法，确保无论是否发生异常，会话都能被正确清理：

class BaseHandler(tornado.web.RequestHandler, SessionMixin):
  def on_exception(self, exception):
    # 异常时回滚活跃会话
    if hasattr(self, 'session') and self.session.is_active:
      self.session.rollback()
    super().on_exception(exception)

  def on_finish(self):
    # 请求结束时关闭/移除会话
    if hasattr(self, 'session'):
      if self.session.is_active:
        self.session.close()
      else:
        self.session.remove()
    super().on_finish()

# 所有页面继承该基类
class MyPage(BaseHandler):
  executor = ThreadPoolExecutor(max_workers=10)
  # ... 业务代码

4. 避免跨线程共享会话 #

RequestHandler的self.session绑定在主线程，后台线程复用会导致连接状态混乱，必须为每个后台任务创建独立会话。

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