标签：Sqlalchemy Flask 数据库 连接 session 进程 sqlalchemy 多线程 options
原创作者：flowell，转载请标明出处：https://www.cnblogs.com/flowell/p/multiprocessing_flask_sqlalchemy.html
Sqlalchemy
flask-sqlalchemy的session是线程安全的，但在多进程环境下，要确保派生子进程时，父进程不存在任何的数据库连接，可以通过调用db.get_engine(app=app).dispose()来手动销毁已经创建的engine，然后再派生子进程。
最近线上的项目总是会报出数据库连接相关的错误，比如“Command out of Sync”，“Mysql server has gone away”，“Lost databse connection”，“Package sequence out of order”等等，最终解决下来，发现以上错误可以分为两种，一种是和连接丢失有关的，一种是和连接被多个线程（进程）同时使用了有关。
我们项目基于flask，有多线程的场景，也有多进程的场景。orm用的是flask的拓展flask-sqlalchemy。flask-sqlalchemy的使用必须基于flask的app实例，也就是说要在app上下文中才能使用flask-sqlalchemy，所以在某些离线（非web）场景下，我们也用到了原生的Sqlalchemy。
原生的Sqlalchemy的使用方式是
engine = create_engine(db_url) Session = sessionmaker(bind=engine) session = Session()
session.query(xxx)
首先要创建一个engine，engine顾名思义就是和数据库连接的引擎。在实际发起查询前，是不会创建任何connection的。创建engine时可以通过指定poolclass参数来指定engine使用的连接池。默认是QueuePool，也可以设置为NullPool（不使用连接池）。为了方便理解，可以把engine视为管理连接池的对象。
sqlalchemy中session和我们平时数据库里说的session是两个不同的概念，在平时数据库中，session的生命周期从连接上数据库开始，到断开和数据库的连接位置。但是sqlalchemy中的session更多的是一种管理连接的对象，它从连接池取出一个连接，使用连接，然后释放连接，而自身也跟随着销毁。sqlalchemy中的Connection对象是管理真正数据库连接的对象，真正的数据库连接在sqlalchemy中是DBAPI。
默认地，如果不传入poolclass，则使用QueuePool（具有一定数量的连接池），如果不指定pool_recycle参数，则默认数据库连接不会刷新。也就是说连接如果不适用，则一直不去刷新它。但是问题来了，在Mysql中，输入“show variables like "%timeout%"; ” ，可以看到有一个waittimeout，还有interacttimeout，默认值为28800（8小时），这两个值代表着，如果8个小时内某个数据库连接都不和mysql联系，那么就会断掉这个连接。所以，8个小时过去了，Mysql把连接断掉了，但是sqlalchemy客户端这边却还保持着这个连接。当某个时候该连接从连接池被取出使用时，就会抛出“Mysql server has gone away”等连接丢失的信息。
解决这个问题的办法很简单，只要传入pool_recycle参数即可。特别地，在flask-sqlalchemy中不会出现这种问题，因为falsk-sqlalchemy拓展自动地帮我们注入了pool_recycle参数，默认为7200秒。
def apply_driver_hacks(self, app, sa_url, options):
"""This method is called before engine creation and used to inject
driver specific hacks into the options. The `options` parameter is
a dictionary of keyword arguments that will then be used to call
the :func:`sqlalchemy.create_engine` function.
The default implementation provides some saner defaults for things
like pool sizes for MySQL and sqlite. Also it injects the setting of
`SQLALCHEMY_NATIVE_UNICODE`.
"""
if sa_url.drivername.startswith('mysql'):
sa_url.query.setdefault('charset', 'utf8')
if sa_url.drivername != 'mysql+gaerdbms':
options.setdefault('pool_size', 10)
options.setdefault('pool_recycle', 7200) # 默认7200秒刷新连接
elif sa_url.drivername == 'sqlite':
pool_size = options.get('pool_size')
detected_in_memory = False
if sa_url.database in (None, '', ':memory:'):
detected_in_memory = True
from sqlalchemy.pool import StaticPool
options['poolclass'] = StaticPool
if 'connect_args' not in options:
options['connect_args'] = {}
options['connect_args']['check_same_thread'] = False
# we go to memory and the pool size was explicitly set
# to 0 which is fail. Let the user know that
if pool_size == 0:
raise RuntimeError('SQLite in memory database with an '
'empty queue not possible due to data '
'loss.')
# if pool size is None or explicitly set to 0 we assume the
# user did not want a queue for this sqlite connection and
# hook in the null pool.
elif not pool_size:
from sqlalchemy.pool import NullPool
options['poolclass'] = NullPool
# if it's not an in memory database we make the path absolute.
if not detected_in_memory:
sa_url.database = os.path.join(app.root_path, sa_url.database)
unu = app.config['SQLALCHEMY_NATIVE_UNICODE']
if unu is None:
unu = self.use_native_unicode
if not unu:
options['use_native_unicode'] = False
if app.config['SQLALCHEMY_NATIVE_UNICODE'] is not None:
warnings.warn(
"The 'SQLALCHEMY_NATIVE_UNICODE' config option is deprecated and will be removed in"
" v3.0. Use 'SQLALCHEMY_ENGINE_OPTIONS' instead.",
DeprecationWarning
)
if not self.use_native_unicode:
warnings.warn(
"'use_native_unicode' is deprecated and will be removed in v3.0."
" Use the 'engine_options' parameter instead.",
DeprecationWarning
)
sessionmaker是Session定制方法，我们把engine传入sessionmaker中，就可以得到一个session工厂，通过工厂来生产真正的session对象。但是这种生产出来的session是线程不安全的，sqlalchemy提供了scoped_session来帮助我们生产线程安全的session，原理类似于Local，就是代理session，通过线程的id来找到真正属于本线程的session。
flask-sqlalchemy就是使用了scoped_session来保证线程安全，具体的代码可以在Sqlalchemy中看到，构造session时，使用了scoped_session。
def create_scoped_session(self, options=None):
"""Create a :class:`~sqlalchemy.orm.scoping.scoped_session`
on the factory from :meth:`create_session`.
An extra key ``'scopefunc'`` can be set on the ``options`` dict to
specify a custom scope function. If it's not provided, Flask's app
context stack identity is used. This will ensure that sessions are
created and removed with the request/response cycle, and should be fine
in most cases.
:param options: dict of keyword arguments passed to session class in
``create_session``
"""
if options is None:
options = {}
scopefunc = options.pop('scopefunc', _app_ctx_stack.__ident_func__)
options.setdefault('query_cls', self.Query)
return orm.scoped_session(
self.create_session(options), scopefunc=scopefunc
)
def create_session(self, options):
"""Create the session factory used by :meth:`create_scoped_session`.
The factory **must** return an object that SQLAlchemy recognizes as a session,
or registering session events may raise an exception.
Valid factories include a :class:`~sqlalchemy.orm.session.Session`
class or a :class:`~sqlalchemy.orm.session.sessionmaker`.
The default implementation creates a ``sessionmaker`` for :class:`SignallingSession`.
:param options: dict of keyword arguments passed to session class
"""
return orm.sessionmaker(class_=SignallingSession, db=self, **options)
多进程和数据库连接
多进程环境下，要注意和数据库连接相关的操作。
说到多进程，python里最常用的就是multiprocessing。multiprocessing在windows下和linux的表现有所区别，在此只讨论linux下的表现。linux下多进程通过fork()来派生，要理解我下面说的必须先弄懂fork()是什么东西。粗略地说，每个进程都有自己的一个空间，称为进程空间，每个进程的进程空间都是独立的，进程与进程之间互不干扰。fork()的作用，就是将一个进程的进程空间，完完全全地copy一份，copy出来的就是子进程了，所以我们说子进程和父进程有着一模一样的地址空间。地址空间就是进程运行的空间，这空间里会有进程已经打开的文件描述符，文件描述符会间接地指向进程已经打开的文件。也就是说，fork()之后，父进程，子进程会有相同的文件描述符，指向相同的一个文件。为什么？因为文件是存在硬盘里的，fork()时copy的内存中的进程空间，并没有把文件也copy一份。这就导致了，父进程，子进程，同时指向同一个文件，他们任意一个都可以对这个文件进行操作。这和本文说的数据库有啥关系？顺着这个思路想，数据库连接是不是一个TCP连接？TCP连接是不是一个socket？socket在linux下是什么，就是一个文件。所以说，如果父进程在fork()之前打开了数据库连接，那么子进程也会拥有这个打开的连接。
两个进程同时写一个连接会导致数据混乱，所以会出现“Command out of sync”的错误，两个进程同时读一个连接，会导致一个进程读到了，另一个没读到，就是“No result”。一个进程关闭了连接，另一个进程并不知道，它试图去操作连接时，就会出现“Lost database connection”的错误。
在此讨论的场景是，父进程在派生子进程之前，父进程拥有已打开的数据库连接。派生出子进程之后，子进程也就拥有了相应的连接。如果在fork()之前父进程没有打开数据库连接，那么也不用担心这个问题。比如Celery使用的prefork池，虽然是多进程模型，但是celery在派子进程前时不会打开数据库连接的，所以不用担心在celery任务中会出现数据库连接混乱的问题。
我做的项目里的多进程的场景之一就是使用tornado来跑web应用，在派生多个web应用实例时，确保此前创建的数据库连接被销毁。
app = Flask() db = Sqlalchemy() db.init_app(app) ... ... db.get_engine(app=app).dispose() # 先销毁已有的engine，确保父进程没有数据库连接 ... ... fork() # 派生子进程
# 例如
tornado.start() # 启动多个web实例进程
标签：Sqlalchemy,Flask,数据库,连接,session,进程,sqlalchemy,多线程,options 来源： https://www.cnblogs.com/flowell/p/multiprocessing_flask_sqlalchemy.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。