mysql插入性能测试

import mysql.connector
import time

# 连接数据库
db = mysql.connector.connect(
    host="localhost",
    user="root",
    password="password",
    database="testdb"
)

cursor = db.cursor()

# 创建测试表
cursor.execute("CREATE TABLE IF NOT EXISTS test_table (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255))")

# 插入单条记录
start_time = time.time()
cursor.execute("INSERT INTO test_table (name) VALUES ('Test')")
end_time = time.time()
print(f"单条插入时间: {end_time - start_time} 秒")

# 批量插入记录
start_time = time.time()
sql = "INSERT INTO test_table (name) VALUES (%s)"
val = [("Test" + str(i)) for i in range(1000)]
cursor.executemany(sql, val)
end_time = time.time()
print(f"批量插入时间: {end_time - start_time} 秒")

# 并发插入记录
def insert_concurrently(num_threads):
    threads = []
    for i in range(num_threads):
        thread = threading.Thread(target=insert_data)
        threads.append(thread)
        thread.start()
    for thread in threads:
        thread.join()

def insert_data():
    cursor.execute("INSERT INTO test_table (name) VALUES ('Concurrent Test')")

start_time = time.time()
insert_concurrently(10)
end_time = time.time()
print(f"并发插入时间: {end_time - start_time} 秒")

# 关闭连接
cursor.close()
db.close()