2022年 11月 6日

python多线程并发访问&数据库连接池原理以及代码

现状

在工作中难免会使用数据库，为了能够高效并发访问数据库，数据库连接池必不可少，由于本站copy模式盛行，导致数据库连接池被错误使用，遇到错误甚至追求能跑通就行。

本文就python版本的数据库链接池模块在实际使用场景样例，来说明如何正确合理的使用数据库连接池。

业务场景 

在部署机器学习模型时采用的是flask框架，模型预测本身是一个很快的事情，无奈有太多的特征需要通过接口（或者是ots，mysql等）获取，导致响应时效性降低。

为了能很好的实现并发性，提升QPS，采用gunicorn进行多进程，异步处理方案。

此时单个进程只有一个数据库链接，就会导致异步执行的线程共用同一个连接，从而导致报错，引入数据库连接池是必须的。

数据库连接池原理

通过预先建立链接，放到然后list中，使用的时候，从list中取出一个链接，使用使用完成后归还连接。当线程太多，链接池中没有链接的时候，可以选择block，等到有链接可用的时候返回，或者是直接返回错误。

dbutils已经实现了两种pooldb：

PooledDB ：可以被多线程共享的链接，适用于异步场景，不断有新线程进来获取连接池，本文使用该方案。
PersistentDB：下面这句话表示，对线程的要求是持续稳定的，不能产生新的线程。

Measures are taken to make the database connections thread-affine.
This means the same thread always uses the same cached connection,
and no other thread will use it.  So even if the underlying DB-API module
is not thread-safe at the connection level this will be no problem here.

For best performance, the application server should keep threads persistent.

dbutils结构如下

db结尾的是mysql等数据库专用。

pg结尾的是PostgreSQL专用。

如上交代完之后，相信你对数据库链接池有较为全面的认知了，好了具体实现代码如下：

主要代码框架逻辑：

1、初始化连接池

2、获取链接

3、查询数据库

4、close链接（返回给连接池，并不是真正的关闭连接池）

5、具体数据查询&解析逻辑根据业务修改，此处提供了sql_fetch_json函数，返回json格式数据。

6、test1为多线程测试，此处自己多运行体会查询结果。

  1. # coding=utf-8
  2. import random
  3.  
  4. import threading
  5. from dbutils.pooled_db import PooledDB
  6. from dbutils.persistent_db import PersistentDB
  7.  
  8. import time
  9. import pymysql
  10.  
  11. from configuration.config import system_logger, db_config
  12.  
  13.  
  14. class MysqlHelper(object):
  15.     def __init__(self, db_config):
  16.         self.__pool = PooledDB(creator=pymysql,
  17.                                mincached=1,
  18.                                maxcached=5,
  19.                                maxshared=5,
  20.                                maxconnections=5,
  21.                                maxusage=5,
  22.                                blocking=True,
  23.                                user=db_config.get('user'),
  24.                                passwd=db_config.get('password'),
  25.                                db=db_config.get('database'),
  26.                                host=db_config.get('host'),
  27.                                port=db_config.get('port'),
  28.                                charset=db_config.get('charset'),
  29.                                )
  30.  
  31.     def getConn(self):
  32.         conn = self.__pool.connection()  # 从连接池获取一个链接
  33.         cursor = conn.cursor()
  34.         return conn, cursor
  35.  
  36.  
  37.     @staticmethod
  38.     def dispose(cursor, conn):
  39.         cursor.close()
  40.         conn.close()
  41.  
  42.     def getOne(self, sql):
  43.  
  44.         conn, cursor = self.getConn()
  45.     
  46.         th_name = threading.currentThread().getName()
  47.         # print(f'{th_name} {self.conn} {self.cursor} {time.time():.4f} start {sql}')
  48.         cursor.execute(sql)
  49.         rows = cursor.fetchall()
  50.         print(f"{th_name} {conn} {cursor} {time.time():.4f} {rows}")
  51.         # self.dispose()
  52.         self.dispose(cursor, conn)
  53.         return rows
  54.  
  55.     def queryOne(self, sql):
  56.         system_logger.info("----------------------sql start ----------------------")
  57.         system_logger.info(sql)
  58.         try:
  59.             conn, cursor = self.getConn()
  60.             result = cursor.execute(sql)
  61.             # rows = cursor.fetchall()
  62.             json_data = self.sql_fetch_json(cursor)
  63.             # 将连接返回
  64.             self.dispose(cursor, conn)
  65.             system_logger.info(f"-----------------------queryByKey result:{result} " + str(json_data))
  66.             if len(json_data) == 1:
  67.                 return json_data[0]
  68.             return None
  69.  
  70.         except Exception as e:
  71.             system_logger.info("-----------predict exception line: " + str(e.__traceback__.tb_lineno) + " of " +
  72.                                e.__traceback__.tb_frame.f_globals["__file__"])
  73.             system_logger.info(e)
  74.             return None
  75.  
  76.     @staticmethod
  77.     def sql_fetch_json(cursor: pymysql.cursors.Cursor):
  78.         """ Convert the pymysql SELECT result to json format """
  79.         keys = []
  80.         for column in cursor.description:
  81.             keys.append(column[0])
  82.         key_number = len(keys)
  83.  
  84.         json_data = []
  85.         for row in cursor.fetchall():
  86.             item = dict()
  87.             for q in range(key_number):
  88.                 item[keys[q]] = row[q]
  89.             json_data.append(item)
  90.         return json_data
  91.  
  92.  
  93. def test1(pool):
  94.     phone_no = f"1390709000{random.randint(6,7)}"
  95.     strsql = f"select * from zy_phone where policy_holder_phone_no={phone_no} order by insure_date " \
  96.              + "desc, kafka_etl_time asc limit 1 "
  97.  
  98.     while True:
  99.         time.sleep(1)
  100.         pool.getOne(strsql)
  101.         # time.sleep(0.001)
  102.         j = 0
  103.         th_name = threading.currentThread().getName()
  104.         # if th_name in ['Thread-2','Thread-5']:
  105.         #     # print(f"task {th_name}")
  106.         #     time.sleep(0.003)
  107.  
  108.  
  109. def main(pool):
  110.     # pool.getConn()
  111.     ths = []
  112.     for i in range(5):
  113.         th = threading.Thread(target=test1, args=(pool,))
  114.         ths.append(th)
  115.     for th in ths:
  116.         th.start()
  117.  
  118.     for th in ths:
  119.         th.join()
  120.  
  121.  
  122. if __name__ == "__main__":
  123.     mysqlhelper = MysqlHelper(db_config)
  124.     main(mysqlhelper)
  125.     time.sleep(3)
  126.     while True:
  127.         time.sleep(1)

常见错误使用方法1：

 def getConn(self):
      self.conn = self.__pool.connection()
      self.cursor = self.conn.cursor() 
此处不应该共享链接，和cursor，会导致报错：

AttributeError: ‘NoneType’ object has no attribute ‘read’

或者：

AttributeError: ‘NoneType’ object has no attribute ‘settimeout‘

常见错误使用方法2：

获取链接以及查询的时候加锁

lock.acquire()
pool.getConn()
pool.getOne(strsql)
lock.release()
time.sleep(1)

因为pooldb本身就会加锁，参见如下源码中，自己在从链接池获取链接,到cursor获取数据的时候加锁，会导致锁冗余，此时连接池会退化成单个数据库链接。

self.__pool.connection() 逻辑如下：

  1.     def connection(self, shareable=True):
  2.         """Get a steady, cached DB-API 2 connection from the pool.
  4.         If shareable is set and the underlying DB-API 2 allows it,
  5.         then the connection may be shared with other threads.
  6.         """
  7.         if shareable and self._maxshared:
  8.             with self._lock:
  9.                 while (not self._shared_cache and self._maxconnections
  10.                         and self._connections >= self._maxconnections):
  11.                     self._wait_lock()
  12.                 if len(self._shared_cache) < self._maxshared:
  13.                     # shared cache is not full, get a dedicated connection
  14.                     try:  # first try to get it from the idle cache
  15.                         con = self._idle_cache.pop(0)
  16.                     except IndexError:  # else get a fresh connection
  17.                         con = self.steady_connection()
  18.                     else:
  19.                         con._ping_check()  # check this connection
  20.                     con = SharedDBConnection(con)
  21.                     self._connections += 1
  22.                 else:  # shared cache full or no more connections allowed
  23.                     self._shared_cache.sort()  # least shared connection first
  24.                     con = self._shared_cache.pop(0)  # get it
  25.                     while con.con._transaction:
  26.                         # do not share connections which are in a transaction
  27.                         self._shared_cache.insert(0, con)
  28.                         self._wait_lock()
  29.                         self._shared_cache.sort()
  30.                         con = self._shared_cache.pop(0)
  31.                     con.con._ping_check()  # check the underlying connection
  32.                     con.share()  # increase share of this connection
  33.                 # put the connection (back) into the shared cache
  34.                 self._shared_cache.append(con)
  35.                 self._lock.notify()
  36.             con = PooledSharedDBConnection(self, con)
  37.         else:  # try to get a dedicated connection
  38.             with self._lock:
  39.                 while (self._maxconnections
  40.                         and self._connections >= self._maxconnections):
  41.                     self._wait_lock()
  42.                 # connection limit not reached, get a dedicated connection
  43.                 try:  # first try to get it from the idle cache
  44.                     con = self._idle_cache.pop(0)
  45.                 except IndexError:  # else get a fresh connection
  46.                     con = self.steady_connection()
  47.                 else:
  48.                     con._ping_check()  # check connection
  49.                 con = PooledDedicatedDBConnection(self, con)
  50.                 self._connections += 1
  51.         return con

到此本文结束，如果觉得有收获，就点个赞吧。