# coding=utf-8
import random
 
import threading
from dbutils.pooled_db import PooledDB
from dbutils.persistent_db import PersistentDB
 
import time
import pymysql
 
from configuration.config import system_logger, db_config
 
 
class MysqlHelper(object):
    def __init__(self, db_config):
        self.__pool = PooledDB(creator=pymysql,
                               mincached=1,
                               maxcached=5,
                               maxshared=5,
                               maxconnections=5,
                               maxusage=5,
                               blocking=True,
                               user=db_config.get('user'),
                               passwd=db_config.get('password'),
                               db=db_config.get('database'),
                               host=db_config.get('host'),
                               port=db_config.get('port'),
                               charset=db_config.get('charset'),
                               )
 
    def getConn(self):
        conn = self.__pool.connection()  # 从连接池获取一个链接
        cursor = conn.cursor()
        return conn, cursor
 
 
    @staticmethod
    def dispose(cursor, conn):
        cursor.close()
        conn.close()
 
    def getOne(self, sql):
 
        conn, cursor = self.getConn()
    
        th_name = threading.currentThread().getName()
        # print(f'{th_name} {self.conn} {self.cursor} {time.time():.4f} start {sql}')
        cursor.execute(sql)
        rows = cursor.fetchall()
        print(f"{th_name} {conn} {cursor} {time.time():.4f} {rows}")
        # self.dispose()
        self.dispose(cursor, conn)
        return rows
 
    def queryOne(self, sql):
        system_logger.info("----------------------sql start ----------------------")
        system_logger.info(sql)
        try:
            conn, cursor = self.getConn()
            result = cursor.execute(sql)
            # rows = cursor.fetchall()
            json_data = self.sql_fetch_json(cursor)
            # 将连接返回
            self.dispose(cursor, conn)
            system_logger.info(f"-----------------------queryByKey result:{result} " + str(json_data))
            if len(json_data) == 1:
                return json_data[0]
            return None
 
        except Exception as e:
            system_logger.info("-----------predict exception line: " + str(e.__traceback__.tb_lineno) + " of " +
                               e.__traceback__.tb_frame.f_globals["__file__"])
            system_logger.info(e)
            return None
 
    @staticmethod
    def sql_fetch_json(cursor: pymysql.cursors.Cursor):
        """ Convert the pymysql SELECT result to json format """
        keys = []
        for column in cursor.description:
            keys.append(column[0])
        key_number = len(keys)
 
        json_data = []
        for row in cursor.fetchall():
            item = dict()
            for q in range(key_number):
                item[keys[q]] = row[q]
            json_data.append(item)
        return json_data
 
 
def test1(pool):
    phone_no = f"1390709000{random.randint(6,7)}"
    strsql = f"select * from zy_phone where policy_holder_phone_no={phone_no} order by insure_date " \
             + "desc, kafka_etl_time asc limit 1 "
 
    while True:
        time.sleep(1)
        pool.getOne(strsql)
        # time.sleep(0.001)
        j = 0
        th_name = threading.currentThread().getName()
        # if th_name in ['Thread-2','Thread-5']:
        #     # print(f"task {th_name}")
        #     time.sleep(0.003)
 
 
def main(pool):
    # pool.getConn()
    ths = []
    for i in range(5):
        th = threading.Thread(target=test1, args=(pool,))
        ths.append(th)
    for th in ths:
        th.start()
 
    for th in ths:
        th.join()
 
 
if __name__ == "__main__":
    mysqlhelper = MysqlHelper(db_config)
    main(mysqlhelper)
    time.sleep(3)
    while True:
        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() 逻辑如下：

    def connection(self, shareable=True):
        """Get a steady, cached DB-API 2 connection from the pool.
        If shareable is set and the underlying DB-API 2 allows it,
        then the connection may be shared with other threads.
        """
        if shareable and self._maxshared:
            with self._lock:
                while (not self._shared_cache and self._maxconnections
                        and self._connections >= self._maxconnections):
                    self._wait_lock()
                if len(self._shared_cache) < self._maxshared:
                    # shared cache is not full, get a dedicated connection
                    try:  # first try to get it from the idle cache
                        con = self._idle_cache.pop(0)
                    except IndexError:  # else get a fresh connection
                        con = self.steady_connection()
                    else:
                        con._ping_check()  # check this connection
                    con = SharedDBConnection(con)
                    self._connections += 1
                else:  # shared cache full or no more connections allowed
                    self._shared_cache.sort()  # least shared connection first
                    con = self._shared_cache.pop(0)  # get it
                    while con.con._transaction:
                        # do not share connections which are in a transaction
                        self._shared_cache.insert(0, con)
                        self._wait_lock()
                        self._shared_cache.sort()
                        con = self._shared_cache.pop(0)
                    con.con._ping_check()  # check the underlying connection
                    con.share()  # increase share of this connection
                # put the connection (back) into the shared cache
                self._shared_cache.append(con)
                self._lock.notify()
            con = PooledSharedDBConnection(self, con)
        else:  # try to get a dedicated connection
            with self._lock:
                while (self._maxconnections
                        and self._connections >= self._maxconnections):
                    self._wait_lock()
                # connection limit not reached, get a dedicated connection
                try:  # first try to get it from the idle cache
                    con = self._idle_cache.pop(0)
                except IndexError:  # else get a fresh connection
                    con = self.steady_connection()
                else:
                    con._ping_check()  # check connection
                con = PooledDedicatedDBConnection(self, con)
                self._connections += 1
        return con