mysql 数据库压力测试

import mysql.connector
import threading
import time

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

# 测试函数
def test_query():
    conn = mysql.connector.connect(**config)
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM your_table")
    result = cursor.fetchall()
    cursor.close()
    conn.close()

# 压力测试
def stress_test(num_threads, num_queries):
    threads = []
    for _ in range(num_threads):
        thread = threading.Thread(target=test_query)
        threads.append(thread)
        thread.start()
    
    for thread in threads:
        thread.join()

if __name__ == "__main__":
    start_time = time.time()
    stress_test(10, 100)  # 10个线程，每个线程执行100次查询
    end_time = time.time()
    print(f"Total time: {end_time - start_time} seconds")