python测试mysql数据库性能（二）

‘host‘: ‘localhost‘, ‘port‘: 3306, ‘database‘: ‘test‘, ‘user‘: ‘root‘, ‘password‘: ‘1234qwer‘, ‘charset‘: ‘utf8‘ } conn = pymysql.connect(**config) cur = conn.cursor() def timer(fn): def _wrapper(count): start = time.time() fn(count) seconds = time.time() - start print(u"{func}函数每 {count} 条数数据写入耗时 {sec}秒".format(func=fn, count=count, sec=seconds)) return _wrapper # 普通写入 @timer def ordinary_insert(count): sql = "insert into students1 (name, age, sex,id,cellphone,address,score) values (‘tom666‘, ‘66‘, ‘boy‘, ‘10066‘, ‘13900000066‘, ‘shanghai‘, ‘66‘)" for i in range(count): cur.execute(sql) # 批量处理 @timer def many_insert(count): sql = "insert into students1 (name, age, sex,id,cellphone,address,score)values(%s,%s,%s,%s,%s,%s,%s)" loop = count / 20 stus = ((‘tom666‘,‘66‘,‘boy‘,‘10066‘,‘13900000066‘,‘shanghai‘,‘66‘),(‘tom666‘,‘66‘,‘boy‘,‘10066‘,‘13900000066‘,‘shanghai‘,‘66‘), (‘tom666‘, ‘66‘, ‘boy‘, ‘10066‘, ‘13900000066‘, ‘shanghai‘, ‘66‘),(‘tom666‘,‘66‘,‘boy‘,‘10066‘,‘13900000066‘,‘shanghai‘,‘66‘), (‘tom666‘, ‘66‘, ‘boy‘, ‘10066‘, ‘13900000066‘, ‘shanghai‘, ‘66‘),(‘tom666‘,‘66‘,‘boy‘,‘10066‘,‘13900000066‘,‘shanghai‘,‘66‘), (‘tom666‘, ‘66‘, ‘boy‘, ‘10066‘, ‘13900000066‘, ‘shanghai‘, ‘66‘),(‘tom666‘,‘66‘,‘boy‘,‘10066‘,‘13900000066‘,‘shanghai‘,‘66‘), (‘tom666‘, ‘66‘, ‘boy‘, ‘10066‘, ‘13900000066‘, ‘shanghai‘, ‘66‘),(‘tom666‘,‘66‘,‘boy‘,‘10066‘,‘13900000066‘,‘shanghai‘,‘66‘), (‘tom666‘, ‘66‘, ‘boy‘, ‘10066‘, ‘13900000066‘, ‘shanghai‘, ‘66‘),(‘tom666‘,‘66‘,‘boy‘,‘10066‘,‘13900000066‘,‘shanghai‘,‘66‘), (‘tom666‘, ‘66‘, ‘boy‘, ‘10066‘, ‘13900000066‘, ‘shanghai‘, ‘66‘),(‘tom666‘,‘66‘,‘boy‘,‘10066‘,‘13900000066‘,‘shanghai‘,‘66‘), (‘tom666‘, ‘66‘, ‘boy‘, ‘10066‘, ‘13900000066‘, ‘shanghai‘, ‘66‘),(‘tom666‘,‘66‘,‘boy‘,‘10066‘,‘13900000066‘,‘shanghai‘,‘66‘), (‘tom666‘, ‘66‘, ‘boy‘, ‘10066‘, ‘13900000066‘, ‘shanghai‘, ‘66‘),(‘tom666‘,‘66‘,‘boy‘,‘10066‘,‘13900000066‘,‘shanghai‘,‘66‘), (‘tom666‘, ‘66‘, ‘boy‘, ‘10066‘, ‘13900000066‘, ‘shanghai‘, ‘66‘),(‘tom666‘,‘66‘,‘boy‘,‘10066‘,‘13900000066‘,‘shanghai‘,‘66‘)) for i in range(int(loop)): cur.executemany(sql, stus) # 事务处理 @timer def transaction_insert(count): sql = "insert into students1 (name, age, sex,id,cellphone,address,score)values(%s,%s,%s,%s,%s,%s,%s)" stus = (‘tom666‘, ‘66‘, ‘boy‘, ‘10066‘, ‘13900000066‘, ‘shanghai‘, ‘66‘) if count > 0: try: for i in range(count): cur.execute(sql, stus) except Exception as e: conn.rollback() # 事务回滚 print(‘事务处理失败‘, e) else: conn.commit() # 事务提交 print(‘事务处理成功, 关闭连接‘, cur.rowcount) cur.close() conn.close() else: print("输入的count有问题，无法执行数据库操作！") def test_insert(count): ordinary_insert(count) many_insert(count) transaction_insert(count) test_insert(20)

输出结果：

E:\python_projects\practises\venv\Scripts\python.exe E:/python_projects/practises/practise20191116/p20191208.py
<function ordinary_insert at 0x0000026994A7BC18>函数每20条数数据写入耗时0.003995656967163086秒
<function many_insert at 0x0000026994A7B8B8>函数每20条数数据写入耗时0.0009996891021728516秒
事务处理成功, 关闭连接 1
<function transaction_insert at 0x0000026994A7BA68>函数每20条数数据写入耗时0.007994651794433594秒

Process finished with exit code 0

 

python测试mysql数据库性能（二）

标签：问题   ini   tab   any   core   事务处理   函数   nec   批量