mysql测试模拟数据处理

import random
import mysql.connector

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

# 创建测试数据
data = []
for i in range(10000):
    data.append((random.randint(1, 100), f"name_{i}", random.randint(1, 1000)))

# 导入数据到MySQL
cursor.executemany("INSERT INTO users (id, name, age) VALUES (%s, %s, %s)", data)
db.commit()

cursor.close()
db.close()

import threading
import mysql.connector

def insert_data():
    db = mysql.connector.connect(host="localhost", user="user", password="password", database="testdb")
    cursor = db.cursor()
    cursor.execute("INSERT INTO users (name, age) VALUES ('test', 25)")
    db.commit()
    cursor.close()
    db.close()

threads = []
for i in range(100):
    thread = threading.Thread(target=insert_data)
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

db = mysql.connector.connect(host="localhost", user="user", password="password", database="testdb")
cursor = db.cursor()

# 查询数据
cursor.execute("SELECT * FROM users WHERE age > 20")
result = cursor.fetchall()

# 验证结果
for row in result:
    print(row)

cursor.close()
db.close()