Flask解析(二):Flask-Sqlalchemy与多线程、多进程

时间:2024-01-25 18:39:50

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实例进程