一、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