mysql集群压力测试

import mysql.connector
import threading
import time

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

# 压力测试函数
def stress_test(num_queries):
    conn = mysql.connector.connect(**config)
    cursor = conn.cursor()
    start_time = time.time()
    for _ in range(num_queries):
        cursor.execute("SELECT * FROM your_table")
    end_time = time.time()
    cursor.close()
    conn.close()
    print(f"Completed {num_queries} queries in {end_time - start_time} seconds")

# 并发线程数
num_threads = 10
# 每个线程执行的查询数
num_queries_per_thread = 1000

threads = []
for _ in range(num_threads):
    thread = threading.Thread(target=stress_test, args=(num_queries_per_thread,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()