MariaDB的线程及连接

简介：

相对于最新的MySQL5.6，MariaDB在性能、功能、管理、NoSQL扩展方面包含了更丰富的特性。比如微秒的支持、线程池、子查询优化、组提交、进度报告等。

本文就主要探索MariaDB当中连接池的一些特性，配置。来配合我们的sqlalchemy。

一：起因

本来是不会写这个东西的，但是，写好了python--flask程序，使用sqlalchemy+mariadb，部署以后总是出问题，500错误之类的。

错误提示是:

raise errors.OperationalError("MySQL Connection not available.")
sqlalchemy.exc.OperationalError: (mysql.connector.errors.OperationalError)
MySQL Connection not available.
[SQL: ‘SELECT ******* \nFROM user‘]
[parameters: [{}]]
(Background on this error at: http://sqlalche.me/e/e3q8)

http://sqlalche.me/e/e3q8:

OperationalError:

Exception raised for errors that are related to the database’s operation andnot necessarily under the control of the programmer, e.g. an unexpecteddisconnect occurs, the data source name is not found, a transaction could notbe processed, a memory allocation error occurred during processing, etc.

This error is aDBAPI Errorand originates fromthe database driver (DBAPI), not SQLAlchemy itself.

TheOperationalErroris the most common (but not the only) error class usedby drivers in the context of the database connection being dropped, or notbeing able to connect to the database. For tips on how to deal with this, seethe sectionDealing with Disconnects.

意思是没有正确断开和数据库的连接。

二:处理断开

http://docs.sqlalchemy.org/en/latest/core/pooling.html#pool-disconnects

官方给了三种方案来解决这个问题：

1.悲观处理

engine = create_engine("mysql+pymysql://user:pw@host/db", pool_pre_ping=True)

pool_pre_ping=True

表示每次连接从池中检查，如果有错误，监测为断开的状态，连接将被立即回收。

2.自定义悲观的ping

from sqlalchemy import exc
from sqlalchemy import event
from sqlalchemy import select

some_engine = create_engine(...)

@event.listens_for(some_engine, "engine_connect")
def ping_connection(connection, branch):
    if branch:
        # "branch" refers to a sub-connection of a connection,
        # we don‘t want to bother pinging on these.
        return

    # turn off "close with result".  This flag is only used with
    # "connectionless" execution, otherwise will be False in any case
    save_should_close_with_result = connection.should_close_with_result
    connection.should_close_with_result = False

    try:
        # run a SELECT 1.   use a core select() so that
        # the SELECT of a scalar value without a table is
        # appropriately formatted for the backend
        connection.scalar(select([1]))
    except exc.DBAPIError as err:
        # catch SQLAlchemy‘s DBAPIError, which is a wrapper
        # for the DBAPI‘s exception.  It includes a .connection_invalidated
        # attribute which specifies if this connection is a "disconnect"
        # condition, which is based on inspection of the original exception
        # by the dialect in use.
        if err.connection_invalidated:
            # run the same SELECT again - the connection will re-validate
            # itself and establish a new connection.  The disconnect detection
            # here also causes the whole connection pool to be invalidated
            # so that all stale connections are discarded.
            connection.scalar(select([1]))
        else:
            raise
    finally:
        # restore "close with result"
        connection.should_close_with_result = save_should_close_with_result

说实话，没怎么看明白。

像是try一个select 语句，如果没问题就关闭。

3.乐观处理

from sqlalchemy import create_engine, exc
e = create_engine(...)
c = e.connect()

try:
    # suppose the database has been restarted.
    c.execute("SELECT * FROM table")
    c.close()
except exc.DBAPIError, e:
    # an exception is raised, Connection is invalidated.
    if e.connection_invalidated:
        print("Connection was invalidated!")

# after the invalidate event, a new connection
# starts with a new Pool
c = e.connect()
c.execute("SELECT * FROM table")

这个看懂了，try一个select语句，如果无效，就返回Connection was invalidated!，然后开一个新的连接，再去执行select。这个应该写个装饰器，放在每个查询前面。

4.使用连接池回收

from sqlalchemy import create_engine
e = create_engine("mysql://scott:tiger@localhost/test", pool_recycle=3600)

这种方式就比较简单了，在连接参数中写上连接超时时间即可。

5.这是自己看文档找到的方法

from sqlalchemy.pool import QueuePool,NullPool,AssertionPool,StaticPool,SingletonThreadPool,Pool

在sqlalchemy.pool下有已经配置好的连接池，直接使用这些连接池也应该可以。

三：测试

docker run  --restart=always --privileged --name My_mariadb_01 -p 3301:3306 -e MYSQL_ROOT_PASSWORD=123456 -d  mariadb:10.2.13
docker run  --restart=always --privileged --name My_mariadb_02 -p 3302:3306 -e MYSQL_ROOT_PASSWORD=123456 -d  mariadb:10.2.13
docker run  --restart=always --privileged --name My_mariadb_03 -p 3303:3306 -e MYSQL_ROOT_PASSWORD=123456 -d  mariadb:10.2.13
docker run  --restart=always --privileged --name My_mariadb_04 -p 3304:3306 -e MYSQL_ROOT_PASSWORD=123456 -d  mariadb:10.2.13
docker run  --restart=always --privileged --name My_mariadb_05 -p 3305:3306 -e MYSQL_ROOT_PASSWORD=123456 -d  mariadb:10.2.13

为避免因数据库交叉连接，首先开启5个MARIADB

Flask_Plan_01   8801       engine = create_engine(‘mysql+mysqlconnector://plan:plan@mysql/plan‘,)
Flask_Plan_02   8802       engine = create_engine(‘mysql+mysqlconnector://plan:plan@mysql/plan‘, pool_pre_ping=True)
Flask_Plan_03   8803       engine = create_engine(‘mysql+mysqlconnector://plan:plan@mysql/plan‘, poolclass=QueuePool)
Flask_Plan_04   8804       engine = create_engine(‘mysql+mysqlconnector://plan:plan@mysql/plan‘, poolclass=NullPool)
Flask_Plan_05   8805       engine = create_engine(‘mysql+mysqlconnector://plan:plan@mysql/plan‘, pool_recycle=3600)

用这5种连接参数进行连接测试。

如果你愿意，也可以继续开，QueuePool,NullPool,AssertionPool,StaticPool,SingletonThreadPool,Pool，把这几种都测试一下。

四：

五：

六：

七：

八：

九：

十：