python DBUtils 线程池 连接 Postgresql（多线程公用线程池，DB-API : psycopg2）

一、DBUtils

DBUtils 是一套允许线程化 Python 程序可以安全和有效的访问数据库的模块，DBUtils提供两种外部接口： PersistentDB ：提供线程专用的数据库连接，并自动管理连接。 PooledDB ：提供线程间可共享的数据库连接，并自动管理连接。

操作数据库模板：

import datetime
import sys
import os
import configparser
import logging
import psycopg2

from  DBUtils.PooledDB import PooledDB


  

class  DatabaseOperator(object):
    '''
    class for database operator
    '''


    def __init__(self, 
        database_config_path, database_config=None):
        '''
        Constructor
        '''
        self._database_config_path = database_config_path
        
        # load database configuration
        if not database_config :
            self._database_config = self.parse_postgresql_config(database_config_path)
        else:
            self._database_config = database_config
        self._pool = None

    def database_config_empty(self):
        if self._database_config:
            return False
        else:
            return True
        
    def parse_postgresql_config(self, database_config_path=None):
        '''解析pei数据库配置文件
            参数
        ---------
        arg1 : conf_file
                        数据库配置文件路径
            返回值
        --------
        dict
                        解析配置属性dict--config
    
            示例
        --------
                无
       '''
        if database_config_path == None and self._database_config_path != None:
            database_config_path = self._database_config_path
        if not os.path.isfile(database_config_path):
            sys.exit("ERROR: Could not find configuration file: {0}".format(database_config_path))
        parser =  configparser.SafeConfigParser()
        parser.read(database_config_path)
        config = {}
        config['database'] = parser.get('UniMonDB', 'Database')
        config['db_user'] = parser.get('UniMonDB', 'UserName')
        config['db_passwd'] = parser.get('UniMonDB', 'Password')
        config['db_port'] = parser.getint('UniMonDB', 'Port')
        config['db_host'] = parser.get('UniMonDB', 'Servername')
        self._database_config = config
        
        return config  
    
    
    def get_pool_conn(self):
        
        if not self._pool:
            self.init_pgsql_pool()
        return self._pool.connection()
        
    def init_pgsql_pool(self):
        '''利用数据库属性连接数据库
                    参数
                ---------
                arg1 : config
                                数据库配置属性
                    返回值
                --------
                
                    示例
                --------
                        无
        '''
        # 字典config是否为空
        config = self.parse_postgresql_config()
        POSTGREIP = config['db_host']
        POSTGREPORT = config['db_port']
        POSTGREDB = config['database']
        POSTGREUSER = config['db_user']
        POSTGREPASSWD = config['db_passwd']
        try:
            logging.info('Begin to create {0} postgresql pool on：{1}.
'.format(POSTGREIP, datetime.datetime.now()))
            
            pool = PooledDB(
                creator=psycopg2,  # 使用链接数据库的模块mincached
                maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制连接数
                mincached=1,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
                maxcached=4,  # 链接池中最多闲置的链接，0和None不限制
                blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
                maxusage=None,  # 一个链接最多被重复使用的次数，None表示无限制
                setsession=[],  # 开始会话前执行的命令列表。
                host=POSTGREIP,
                port=POSTGREPORT,
                user=POSTGREUSER,
                password=POSTGREPASSWD,
                database=POSTGREDB)
            self._pool = pool    
            logging.info('SUCCESS: create postgresql success.
')
                    
        except Exception as e:
            logging.error('ERROR: create postgresql pool failed：{0}
')
            self.close_db_cursor()
            sys.exit('ERROR: create postgresql pool error caused by {0}'.format(str(e)))

            
    def pg_select_operator(self, sql):
        '''进行查询操作,函数返回前关闭cursor,conn
                    参数
                ---------
                arg1 : sql查询语句
                    返回值
                --------
                list:result
                                        类型为list的查询结果:result
            
                    示例
                --------
                        无
        '''
        # 执行查询
        try:
            conn = self.get_pool_conn()
            cursor = conn.cursor()      
            cursor.execute(sql)
            result = cursor.fetchall()
        except Exception as e:
            logging.error('ERROR: execute  {0} causes error'.format(sql))
            sys.exit('ERROR: load data from database error caused {0}'.format(str(e)))
        finally:
            cursor.close()
            conn.close()       
        return result

    def test_pool_con(self):
        sql = 'select * from tbl_devprofile'
        result = self.pg_select_operator(sql)
        print(result)
        
    def pg_insert_operator(self, sql):
        
        result = False
        try:
            conn = self.get_pool_conn()
            cursor = conn.cursor()      
            cursor.execute(sql)
            result =  True
        except Exception as e:
            logging.error('ERROR: execute  {0} causes error'.format(sql))
            sys.exit('ERROR: insert data from database error caused {0}'.format(str(e)))
        finally:
            cursor.close()
            conn.commit()
            conn.close()    
        return result
    
    def pg_update_operator(self, sql):
        
        result = False
        try:
            conn = self.get_pool_conn()
            cursor = conn.cursor()      
            cursor.execute(sql)
            result =  True
        except Exception as e:
            logging.error('ERROR: execute  {0} causes error'.format(sql))
            sys.exit('ERROR: update data from database error caused {0}'.format(str(e)))
        finally:
            cursor.close()
            conn.commit()
            conn.close()    
        return result

    def pg_delete_operator(self, sql):
        result = False
        # 执行查询
        try:
            conn = self.get_pool_conn()
            cursor = conn.cursor()   
            cursor.execute(sql)
            result =  True
        except Exception as e:
            logging.error('ERROR: execute  {0} causes error'.format(sql))
            sys.exit('ERROR: delete data from database error caused {0}'.format(str(e)))
        finally:
            cursor.close()
            conn.commit()
            conn.close()       
        return result

    
    def close_pool(self):
        '''关闭pool
                    参数
                ---------
                        无 

                    返回值
                --------
                        无
                    示例
                --------
                        无
        '''
        if self._pool != None:
            self._pool.close()
            
if __name__ == '__main__':
    path = "E:\Users\Administrator\eclipse-workspace\com.leagsoft.basemodule\base\config\sql_conf.conf"
    db = DatabaseOperator(
    database_config_path=path)
    db.test_pool_con()

二、多线程

原理：创建多个线程类，多个线程类共享一个队里Queue，每一个线程类可以操作数据库

from threading import Thread
    
class Worker(Thread):
    def __init__(self, queue):
        Thread.__init__(self)
        self.queue = queue
 
    def run(self):
        while True:
            # Get the work from the queue and expand the tuple
            # 从队列中获取任务
            database_operator, device, stand_alone_result = self.queue.get()
            operateResult(database_operator, device, stand_alone_result)
            # 任务执行完之后要通知队列
            self.queue.task_done()

填充队列

    # 使用队列多线程
    logging.info('begin to update all device risk score by multi_processing.
')
    from queue import Queue
    queue = Queue()
    # 六个线程，每个线程共享一个队列
    for _ in range(6):
        worker = Worker(queue)
        worker.setDaemon(True)
        worker.start()
          
    for record in all_devid:
        device = record[0]
        devtype = record[1]
        all_countlist = all_dict.get(device)
        stand_alone_result = device_assess(all_countlist)
        if (devtype in (server_devtype + computer_devtype)) and (stand_alone_result < 100):
            stand_alone_result *= 0.8
        # 将设备风险评分数据保存到数据库中
        queue.put((database_operator, device, stand_alone_result))
     
    #等待队列任务执行完
    queue.join()


def operateResult(database_operator, device, stand_alone_result):
    '''
    函数名称: device_assess
    描述:  保存单台设备分数到数据库
    调用: 无
    被调用:  main
    被访问的表: tbl_devprofile
    被修改的表: 无
    输入参数: database_operator, device:设备uid, stand_alone_result:单台设备风险分数
    输出参数:无
    返回值: 单台设备风险分数值
    其它:  无
    '''
    import time
    find_profile_sql = "SELECT uiddevrecordid FROM tbl_devprofile WHERE uiddevrecordid='{0}';".format(device)
    isExistRecord = database_operator.pg_select_operator(find_profile_sql)
    #currentTime=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    currentTime=time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))
    if len(isExistRecord) > 0:
        updata_profile_sql = "UPDATE tbl_devprofile SET irisklevel={0}, dtrisktime='{1}' 
                              WHERE uiddevrecordid='{2}';".format(stand_alone_result, currentTime, device)
        database_operator.pg_update_operator(updata_profile_sql)
    else:
        insert_profile_sql = "INSERT INTO tbl_devprofile VALUES('{0}',NULL,NULL,NULL,NULL,NULL,NULL,NULL,{1},'{2}');".format(
            device, stand_alone_result, currentTime)
        database_operator.pg_insert_operator(insert_profile_sql)

使用单线程时，执行完代码花费20s左右，使用多线程时花费5s左右。

Reference：

[1] https://blog.csdn.net/zhaihaifei/article/details/54016939

[2] https://www.cnblogs.com/hao-ming/p/7215050.html?utm_source=itdadao&utm_medium=referral

[3] https://www.cnblogs.com/wozijisun/p/6160065.html (多线程)

[4] http://www.lpfrx.com/archives/4431/

[5] https://www.cnblogs.com/95lyj/p/9047554.html