在软件的性能或压力测试中,需要向服务器的数据库持续输入数据,来验证大量IO情况下服务器及应用软件能否正常工作。本文介绍一种向Mysql数据库持续插入图片数据的简单方法。
准备工作
安装pymysql模块
pip install pymysql
源代码
#encoding=utf-8
import pymysql
import time
import string
import random
#图片路径(根据实际情况修改)示例如下:
picUrl = "/test/index.jpg"
#连接的数据库服务器主机名(根据实际情况修改)示例如下:
#conhost="127.0.0.1"
conhost="localhost"
#连接的数据库端口号(根据实际情况修改)示例如下:
conport=3306
#连接的数据库的用户名(根据实际情况修改)示例如下:
conuser="root"
#连接的数据库密码(根据实际情况修改)示例如下:
conpasswd="111111"
#连接的数据库名(根据实际情况修改)示例如下:
condb="test"
class BlobDataTestor:
def __init__(self):
#连接数据库
self.conn = pymysql.connect(host=conhost,port=conport, user=conuser,passwd=conpasswd,db=condb)
def __del__(self):
try:
self.conn.close()
except:
pass
def closedb(self):
self.conn.close()
#创建表
def setup(self):
cursor = self.conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS `Dem_Picture` (
`ID` int(11) NOT NULL auto_increment,
`time` timestamp ,
`PicData` mediumblob,
`RandomNum` varchar(20),
PRIMARY KEY (`ID`)
) ENGINE=MyISAM DEFAULT CHARSET=utf8 AUTO_INCREMENT=1 ;
""")
#删除表
def teardown(self):
cursor = self.conn.cursor()
try:
cursor.execute("Drop Table Dem_Picture")
except:
pass
# self.conn.commit()
def testRWBlobData(self):
# 读取源图片数据
f = open(picUrl, "rb")
b = f.read()
f.close()
# 生成随机数
s = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(10))
# 将数据写入表
cursor = self.conn.cursor()
cursor.execute("INSERT INTO Dem_Picture (time,PicData,RandomNum) VALUES (current_timestamp,%s,%s)",((pymysql.Binary(b)),s))
if __name__ == "__main__":
test = BlobDataTestor()
try:
#创建表
test.setup()
while 1:
test.testRWBlobData()
print "insert success"
#time.sleep(1) #根据需要设置插入时间间隔
#删除表
#test.teardown()
finally:
test.closedb()