1.安装环境

pip install locust mysql-connector-python

2.设置测试环境

打开MySQL服务

打开Navicat新建查询，输入SQL语句

3.编写locust脚本

load_mysql.py

# coding=utf-8
from locust import User, TaskSet, task, between
import mysql.connector
import random


class MySQLTaskSet(TaskSet):
    def __init__(self, parent: User):
        super().__init__(parent)
        self.cursor = None
        self.db = None

    def on_start(self):
        """初始化 MySQL 连接"""
        try:
            self.db = mysql.connector.connect(
                host="localhost",
                user="root",
                password="123456",
                database="testdb"
            )
            self.cursor = self.db.cursor()
            print("MySQL connection established")
        except mysql.connector.Error as err:
            print(f"MySQL connection failed: {err}")

    def on_stop(self):
        """关闭 MySQL 连接"""
        if hasattr(self, 'cursor'):
            self.cursor.close()
        if hasattr(self, 'db'):
            self.db.close()
        print("MySQL connection closed")

    @task(1)
    def write_data(self):
        """执行 INSERT 操作"""
        try:
            name = random.choice(["Alice", "Bob", "Charlie"])
            age = random.randint(20, 50)
            self.cursor.execute(
                "INSERT INTO test_table (name, age) VALUES (%s, %s)",
                (name, age)
            )
            self.db.commit()
            print(f"Inserted row with name={name}, age={age}")
        except Exception as e:
            print(f"Write operation failed: {e}")


class MySQLUser(User):
    tasks = [MySQLTaskSet]
    wait_time = between(1, 5)

4.执行测试

打开终端执行命令

locust -f load_mysql.py --headless -u 3 -r 1 -t 10s

执行结果

打开Navicat查看test_table表可以看到数据增多