mysql 压力测试

import mysql.connector
from concurrent.futures import ThreadPoolExecutor

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

# 测试SQL语句
sql = "SELECT * FROM your_table LIMIT 1"

def execute_query():
    try:
        conn = mysql.connector.connect(**config)
        cursor = conn.cursor()
        cursor.execute(sql)
        result = cursor.fetchone()
        cursor.close()
        conn.close()
        return result
    except Exception as e:
        print(f"Error: {e}")
        return None

if __name__ == "__main__":
    # 并发数
    concurrency = 100
    with ThreadPoolExecutor(max_workers=concurrency) as executor:
        futures = [executor.submit(execute_query) for _ in range(concurrency)]
        results = [future.result() for future in futures]
        print(f"Results: {results}")