locust 压测mysql

Locust 压测 MySQL 需要使用 Python 编写自定义代码来执行 SQL 查询和插入操作，并在 Locust 中调用该代码进行压力测试。

以下是一个示例代码，其中使用了 PyMySQL 库来连接和查询 MySQL 数据库：

import pymysql

class MySQLClient:
    def __init__(self, host, user, password, database):
        self.host = host
        self.user = user
        self.password = password
        self.database = database
        self.conn = None

    def connect(self):
        self.conn = pymysql.connect(
            host=self.host,
            user=self.user,
            password=self.password,
            database=self.database
        )

    def disconnect(self):
        if self.conn is not None:
            self.conn.close()

    def select(self):
        with self.conn.cursor() as cur:
            cur.execute("SELECT * FROM users")
            rows = cur.fetchall()
            return rows

    def insert(self, name, email):
        with self.conn.cursor() as cur:
            cur.execute("INSERT INTO users (name, email) VALUES (%s, %s)", (name, email))
            self.conn.commit()

# Example usage:
# client = MySQLClient("localhost", "root", "password", "test")
# client.connect()
# results = client.select()
# client.insert("test user", "test@example.com")
# client.disconnect()

在 Locust 中，我们可以将此代码集成为 Task，以便可以测试其性能。此处使用 HttpUser 来进行测试：

from locust import HttpUser, TaskSet, task

class MyUser(HttpUser):
    @task
    def mysql_select(self):
        client = MySQLClient("localhost", "root", "password", "test")
        client.connect()
        results = client.select()
        client.disconnect()

    @task
    def mysql_insert(self):
        client = MySQLClient("localhost", "root", "password", "test")
        client.connect()
        client.insert("test user", "test@example.com")
        client.disconnect()

通过运行此测试，我们可以看到在给定的时间内对 MySQL 数据库执行的查询和插入操作的性能指标。需要记住的是，这仅提供了基本的示例代码，实际上还需要根据具体需求编写更详细的测试用例。