一、DBUtils
DBUtils 是一套允许线程化 Python 程序可以安全和有效的访问数据库的模块，DBUtils提供两种外部接口： PersistentDB ：提供线程专用的数据库连接，并自动管理连接。 PooledDB ：提供线程间可共享的数据库连接，并自动管理连接。
操作数据库模板：
1 import datetime 2 import sys 3 import os 4 import configparser 5 import logging 6 import psycopg2 7 8 from DBUtils.PooledDB import PooledDB 9 10 11 12 13 class DatabaseOperator(object): 14 ''' 15 class for database operator 16 ''' 17 18 19 def __init__(self, 20 database_config_path, database_config=None): 21 ''' 22 Constructor 23 ''' 24 self._database_config_path = database_config_path 25 26 # load database configuration 27 if not database_config : 28 self._database_config = self.parse_postgresql_config(database_config_path) 29 else: 30 self._database_config = database_config 31 self._pool = None 32 33 def database_config_empty(self): 34 if self._database_config: 35 return False 36 else: 37 return True 38 39 def parse_postgresql_config(self, database_config_path=None): 40 '''解析pei数据库配置文件 41 参数 42 --------- 43 arg1 : conf_file 44 数据库配置文件路径 45 返回值 46 -------- 47 dict 48 解析配置属性dict--config 49 50 示例 51 -------- 52 无 53 ''' 54 if database_config_path == None and self._database_config_path != None: 55 database_config_path = self._database_config_path 56 if not os.path.isfile(database_config_path): 57 sys.exit("ERROR: Could not find configuration file: {0}".format(database_config_path)) 58 parser = configparser.SafeConfigParser() 59 parser.read(database_config_path) 60 config = {} 61 config['database'] = parser.get('UniMonDB', 'Database') 62 config['db_user'] = parser.get('UniMonDB', 'UserName') 63 config['db_passwd'] = parser.get('UniMonDB', 'Password') 64 config['db_port'] = parser.getint('UniMonDB', 'Port') 65 config['db_host'] = parser.get('UniMonDB', 'Servername') 66 self._database_config = config 67 68 return config 69 70 71 def get_pool_conn(self): 72 73 if not self._pool: 74 self.init_pgsql_pool() 75 return self._pool.connection() 76 77 def init_pgsql_pool(self): 78 '''利用数据库属性连接数据库 79 参数 80 --------- 81 arg1 : config 82 数据库配置属性 83 返回值 84 -------- 85 86 示例 87 -------- 88 无 89 ''' 90 # 字典config是否为空 91 config = self.parse_postgresql_config() 92 POSTGREIP = config['db_host'] 93 POSTGREPORT = config['db_port'] 94 POSTGREDB = config['database'] 95 POSTGREUSER = config['db_user'] 96 POSTGREPASSWD = config['db_passwd'] 97 try: 98 logging.info('Begin to create {0} postgresql pool on：{1}.\n'.format(POSTGREIP, datetime.datetime.now())) 99 100 pool = PooledDB( 101 creator=psycopg2, # 使用链接数据库的模块mincached 102 maxconnections=6, # 连接池允许的最大连接数，0和None表示不限制连接数 103 mincached=1, # 初始化时，链接池中至少创建的空闲的链接，0表示不创建 104 maxcached=4, # 链接池中最多闲置的链接，0和None不限制 105 blocking=True, # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错 106 maxusage=None, # 一个链接最多被重复使用的次数，None表示无限制 107 setsession=[], # 开始会话前执行的命令列表。 108 host=POSTGREIP, 109 port=POSTGREPORT, 110 user=POSTGREUSER, 111 password=POSTGREPASSWD, 112 database=POSTGREDB) 113 self._pool = pool 114 logging.info('SUCCESS: create postgresql success.\n') 115 116 except Exception as e: 117 logging.error('ERROR: create postgresql pool failed：{0}\n') 118 self.close_db_cursor() 119 sys.exit('ERROR: create postgresql pool error caused by {0}'.format(str(e))) 120 121 122 def pg_select_operator(self, sql): 123 '''进行查询操作,函数返回前关闭cursor,conn 124 参数 125 --------- 126 arg1 : sql查询语句 127 返回值 128 -------- 129 list:result 130 类型为list的查询结果:result 131 132 示例 133 -------- 134 无 135 ''' 136 # 执行查询 137 try: 138 conn = self.get_pool_conn() 139 cursor = conn.cursor() 140 cursor.execute(sql) 141 result = cursor.fetchall() 142 except Exception as e: 143 logging.error('ERROR: execute {0} causes error'.format(sql)) 144 sys.exit('ERROR: load data from database error caused {0}'.format(str(e))) 145 finally: 146 cursor.close() 147 conn.close() 148 return result 149 150 def test_pool_con(self): 151 sql = 'select * from tbl_devprofile' 152 result = self.pg_select_operator(sql) 153 print(result) 154 155 def pg_insert_operator(self, sql): 156 157 result = False 158 try: 159 conn = self.get_pool_conn() 160 cursor = conn.cursor() 161 cursor.execute(sql) 162 result = True 163 except Exception as e: 164 logging.error('ERROR: execute {0} causes error'.format(sql)) 165 sys.exit('ERROR: insert data from database error caused {0}'.format(str(e))) 166 finally: 167 cursor.close() 168 conn.commit() 169 conn.close() 170 return result 171 172 def pg_update_operator(self, sql): 173 174 result = False 175 try: 176 conn = self.get_pool_conn() 177 cursor = conn.cursor() 178 cursor.execute(sql) 179 result = True 180 except Exception as e: 181 logging.error('ERROR: execute {0} causes error'.format(sql)) 182 sys.exit('ERROR: update data from database error caused {0}'.format(str(e))) 183 finally: 184 cursor.close() 185 conn.commit() 186 conn.close() 187 return result 188 189 def pg_delete_operator(self, sql): 190 result = False 191 # 执行查询 192 try: 193 conn = self.get_pool_conn() 194 cursor = conn.cursor() 195 cursor.execute(sql) 196 result = True 197 except Exception as e: 198 logging.error('ERROR: execute {0} causes error'.format(sql)) 199 sys.exit('ERROR: delete data from database error caused {0}'.format(str(e))) 200 finally: 201 cursor.close() 202 conn.commit() 203 conn.close() 204 return result 205 206 207 def close_pool(self): 208 '''关闭pool 209 参数 210 --------- 211 无 212 213 返回值 214 -------- 215 无 216 示例 217 -------- 218 无 219 ''' 220 if self._pool != None: 221 self._pool.close() 222 223 if __name__ == '__main__': 224 path = "E:\\Users\\Administrator\\eclipse-workspace\\com.leagsoft.basemodule\\base\\config\\sql_conf.conf" 225 db = DatabaseOperator( 226 database_config_path=path) 227 db.test_pool_con()
二、多线程
原理：创建多个线程类，多个线程类共享一个队里Queue，每一个线程类可以操作数据库
1 from threading import Thread 2 3 class Worker(Thread): 4 def __init__(self, queue): 5 Thread.__init__(self) 6 self.queue = queue 7 8 def run(self): 9 while True: 10 # Get the work from the queue and expand the tuple 11 # 从队列中获取任务 12 database_operator, device, stand_alone_result = self.queue.get() 13 operateResult(database_operator, device, stand_alone_result) 14 # 任务执行完之后要通知队列 15 self.queue.task_done()
填充队列
1 # 使用队列多线程 2 logging.info('begin to update all device risk score by multi_processing.\n') 3 from queue import Queue 4 queue = Queue() 5 # 六个线程，每个线程共享一个队列 6 for _ in range(6): 7 worker = Worker(queue) 8 worker.setDaemon(True) 9 worker.start() 10 11 for record in all_devid: 12 device = record[0] 13 devtype = record[1] 14 all_countlist = all_dict.get(device) 15 stand_alone_result = device_assess(all_countlist) 16 if (devtype in (server_devtype + computer_devtype)) and (stand_alone_result < 100): 17 stand_alone_result *= 0.8 18 # 将设备风险评分数据保存到数据库中 19 queue.put((database_operator, device, stand_alone_result)) 20 21 #等待队列任务执行完 22 queue.join() 23 24 25 def operateResult(database_operator, device, stand_alone_result): 26 ''' 27 函数名称: device_assess 28 描述: 保存单台设备分数到数据库 29 调用: 无 30 被调用: main 31 被访问的表: tbl_devprofile 32 被修改的表: 无 33 输入参数: database_operator, device:设备uid, stand_alone_result:单台设备风险分数 34 输出参数:无 35 返回值: 单台设备风险分数值 36 其它: 无 37 ''' 38 import time 39 find_profile_sql = "SELECT uiddevrecordid FROM tbl_devprofile WHERE uiddevrecordid='{0}';".format(device) 40 isExistRecord = database_operator.pg_select_operator(find_profile_sql) 41 #currentTime=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') 42 currentTime=time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time())) 43 if len(isExistRecord) > 0: 44 updata_profile_sql = "UPDATE tbl_devprofile SET irisklevel={0}, dtrisktime='{1}' \ 45 WHERE uiddevrecordid='{2}';".format(stand_alone_result, currentTime, device) 46 database_operator.pg_update_operator(updata_profile_sql) 47 else: 48 insert_profile_sql = "INSERT INTO tbl_devprofile VALUES('{0}',NULL,NULL,NULL,NULL,NULL,NULL,NULL,{1},'{2}');".format( 49 device, stand_alone_result, currentTime) 50 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
