mysql 如何进行压力测试

import mysql.connector
import threading
import time

# 数据库连接配置
config = {
    'user': 'your_user',
    'password': 'your_password',
    'host': 'your_host',
    'database': 'your_database'
}

# 并发用户数
concurrent_users = 100

# 每个用户的查询次数
queries_per_user = 100

def run_queries(user_id):
    conn = mysql.connector.connect(**config)
    cursor = conn.cursor()
    for _ in range(queries_per_user):
        cursor.execute("SELECT * FROM your_table WHERE id = %s", (user_id % 100,))
        result = cursor.fetchall()
    cursor.close()
    conn.close()

threads = []
start_time = time.time()

for i in range(concurrent_users):
    thread = threading.Thread(target=run_queries, args=(i,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

end_time = time.time()
total_time = end_time - start_time
print(f"Total time: {total_time} seconds")
print(f"Throughput: {concurrent_users * queries_per_user / total_time} queries/second")