带有池的 SQLAlchemy 不会关闭数据库连接

Question

我发现 SQLAlchemy 不会释放数据库连接（在我的情况下），所以这会导致服务器崩溃。连接是由不同的线程构成的。

这里是简化的代码

"""
Test to see DB connection allocation size while making call from multiple threads
"""

from time import sleep
from threading import Thread, current_thread
import uuid

from sqlalchemy import func, or_, desc
from sqlalchemy import event
from sqlalchemy import ForeignKey, Column, Integer, String, DateTime, UniqueConstraint
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import relationship
from sqlalchemy.orm import scoped_session, Session
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.types import Integer, DateTime, String, Boolean, Text, Float
from sqlalchemy.engine import Engine
from sqlalchemy.pool import NullPool

# MySQL
SQLALCHEMY_DATABASE = 'mysql'
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://amalgam:amalgam@localhost/amalgam?charset=utf8mb4' # https://stackoverflow.com/questions/47419943/pymysql-warning-1366-incorrect-string-value-xf0-x9f-x98-x8d-t
SQLALCHEMY_ECHO = False
SQLALCHEMY_ENGINE_OPTIONS = {'pool_size': 40, 'max_overflow': 0}
SQLALCHEMY_ISOLATION_LEVEL = "AUTOCOMMIT"

# DB Engine

# engine = create_engine(SQLALCHEMY_DATABASE_URI, echo=SQLALCHEMY_ECHO, pool_recycle=3600,
#                        isolation_level= SQLALCHEMY_ISOLATION_LEVEL,
#                        **SQLALCHEMY_ENGINE_OPTIONS
#                        ) #  Connect to server

engine = create_engine(SQLALCHEMY_DATABASE_URI, 
                        echo=SQLALCHEMY_ECHO, 
                        # poolclass=NullPool,
                        pool_recycle=3600,
                       isolation_level= SQLALCHEMY_ISOLATION_LEVEL,
                       **SQLALCHEMY_ENGINE_OPTIONS
                       ) #  Connect to server


session_factory = sessionmaker(bind=engine)
Base = declarative_base()

# ORM Entity
class User(Base):

    LEVEL_NORMAL = 'normal'
    LEVEL_ADMIN = 'admin'

    __tablename__ = "users"
    id = Column(Integer, primary_key=True)    
    name = Column(String(100), nullable=True)
    email = Column(String(100), nullable=True, unique=True)
    password = Column(String(100), nullable=True)
    level = Column(String(100), default=LEVEL_NORMAL)


# Workers
NO = 10
workers = []

_scoped_session_factory = scoped_session(session_factory)


def job(job_id):
    session = _scoped_session_factory()

    print("Job is {}".format(job_id))

    user = User(name='User {} {}'.format(job_id, uuid.uuid4()), email='who cares {} {}'.format(job_id, uuid.uuid4()))

    session.add(user)
    session.commit()
    session.close()

    print("Job {} done".format(job_id))
    sleep(10)
    
# Create worker threads
for i in range(NO):
    workers.append(Thread(target=job, kwargs={'job_id':i}))

# Start them
for worker in workers:
    worker.start()

# Join them
for worker in workers:
    worker.join()

# Allow some time to see MySQL's "show processlist;" command
sleep(10)

程序到达的那一刻

sleep(10)

然后我运行

show processlist;

它给出以下结果 - 意味着与数据库的所有连接仍然存在。

如何强制关闭这些连接？

注意：我可以利用

poolclass=NullPool

但我觉得该解决方案限制性太强 - 我希望仍然可以访问数据库池，但能够在需要时以某种方式关闭连接

Answer 1

以下来自signature for QueuePool constructor

pool_size – 要维护的池大小，默认为 5。这 是永久保持的最大连接数 游泳池。请注意，池开始时没有连接；一旦这 请求连接数，该连接数将 保持。 pool_size 可以设置为 0 表示没有大小限制；到 禁用池，改用NullPool。

max_overflow – 池的最大溢出大小。当数 已签出的连接数达到 pool_size 中设置的大小， 额外的连接将返回到此限制。当那些 额外的连接返回到池中，它们被断开 并丢弃。那么同时的总数 池将允许的连接是 pool_size + max_overflow，并且 池将允许的“睡眠”连接总数为 池大小。 max_overflow 可以设置为 -1 表示没有溢出 限制;并发总数不设限制 连接。默认为 10。

SQLALCHEMY_ENGINE_OPTIONS = {'pool_size': 40, 'max_overflow': 0}

鉴于上述情况，此配置要求 SQLAlchemy 保持最多 40 个连接打开。

如果你不喜欢这样，但想保持一些可用的连接，你可以尝试这样的配置：

SQLALCHEMY_ENGINE_OPTIONS = {'pool_size': 10, 'max_overflow': 30}

这将在池中保留 10 个持久连接，如果同时请求，则最多会爆发 40 个连接。任何超过配置的池大小的连接都会在被检回池后立即关闭。