Python 模拟Jmeter对数据库进行并发测试
时间: 2023-08-10 09:03:40 浏览: 172
可以使用 Python 的 multiprocessing 模块来实现多进程并发测试,同时使用 PyMySQL 或者其他 Python 数据库操作库来对数据库进行读写操作。
以下是一个简单的示例代码,实现了同时启动多个进程对数据库进行并发测试:
```python
import pymysql
from multiprocessing import Pool, Process
# 数据库连接信息
db_config = {
'host': 'localhost',
'port': 3306,
'user': 'root',
'password': 'password',
'db': 'test_db',
'charset': 'utf8mb4'
}
# 并发测试的进程数
process_num = 10
# 测试数据
test_data = [('user1', 'password1'), ('user2', 'password2'), ('user3', 'password3')]
# 测试函数,对数据库进行读写操作
def test_db(data):
conn = pymysql.connect(**db_config)
cursor = conn.cursor()
# 插入数据
insert_sql = "insert into user(username, password) values (%s, %s)"
cursor.execute(insert_sql, data)
conn.commit()
# 查询数据
select_sql = "select * from user where username=%s"
cursor.execute(select_sql, (data[0],))
result = cursor.fetchone()
print(result)
cursor.close()
conn.close()
if __name__ == '__main__':
# 使用进程池同时启动多个进程
pool = Pool(processes=process_num)
for data in test_data:
pool.apply_async(test_db, args=(data,))
pool.close()
pool.join()
```
这个示例代码同时启动了 10 个进程,对数据库进行并发读写操作,每个进程都会执行 `test_db` 函数,将对应的测试数据插入到数据库中,并查询出来。可以根据实际需求修改测试数据和测试函数,以适应不同的测试场景。
阅读全文