1.安装环境

安装redis

brew install redis

开启redis服务

brew services start redis

停止redis服务

brew services stop redis

安装Python库

pip install locust redis

2.编写脚本

loadTest.py

# coding=utf-8
import json
import random
import time
import redis
from locust import User, events, task, tag
from typing import Optional

host_name = "localhost"
port_no = 6379


def _execute(command: str, key: str, func, *args, **kwargs) -> Optional[object]:
    """
    通用方法：执行 Redis 操作并触发 Locust 事件
    """
    start_time = time.time()
    try:
        result = func(key, *args, **kwargs)
        total_time = int((time.time() - start_time) * 1000)
        response_length = len(str(result)) if result else 0
        events.request.fire(
            request_type=command,
            name=key,
            response_time=total_time,
            response_length=response_length,
            exception=None
        )
        return result
    except Exception as e:
        total_time = int((time.time() - start_time) * 1000)
        events.request.fire(
            request_type=command,
            name=key,
            response_time=total_time,
            response_length=0,
            exception=e
        )
        return None


class RedisClient:
    def __init__(self, host: str = host_name, port: int = port_no):
        self.rc = redis.StrictRedis(host=host, port=port)

    def get_query_string(self, key: str):
        return _execute("GET", key, self.rc.get)

    def set_query_string(self, key: str, value: Optional[dict] = None):
        value = value or {"bids": random.randint(47238, 57238)}
        return _execute("SET", key, self.rc.set, json.dumps(value))

    def lpush_in_list(self, key: str):
        return _execute("LPUSH", key, self.rc.lpush, 0, 0, 0, 0, 0)

    def sadd_in_set(self, key: str):
        visitors = {"dan", "jon", "alex"}
        return _execute("SADD", key, self.rc.sadd, *visitors)

    def hset_in_hash(self, key: str, field: str = "1", value: str = "One"):
        return _execute("HSET", key, self.rc.hset, field, value)

    def hget_in_hash(self, key: str, field: str = "1"):
        return _execute("HGET", key, self.rc.hget, field)

    def hdel_in_hash(self, key: str, field: str = "1"):
        return _execute("HDEL", key, self.rc.hdel, field)

    def zadd_in_sorted_set(self, key: str, player_name: str = "Player1", score: int = 56):
        return _execute("ZADD", key, self.rc.zadd, {player_name: score}, nx=False)

    def zrange_in_sorted_set(self, key: str):
        return _execute("ZRANGE", key, self.rc.zrange, 0, -1)


class RedisLocust(User):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.client = RedisClient()

    @task
    @tag("string")
    def string_operations(self):
        self.client.set_query_string("string_set_operation")
        self.client.get_query_string("string_get_operation")

    @task
    @tag("list")
    def list_operation(self):
        self.client.lpush_in_list("list_lpush_operation")

    @task
    @tag("set")
    def set_operation(self):
        self.client.sadd_in_set("set_sadd_operation")

    @task
    @tag("hash")
    def hash_operation(self):
        self.client.hset_in_hash("hash_hset_operation")
        self.client.hget_in_hash("hash_hget_operation")
        self.client.hdel_in_hash("hash_hdel_operation")

    @task
    @tag("sorted-set")
    def sorted_set_operation(self):
        self.client.zadd_in_sorted_set("sorted_set_zadd_operation")
        self.client.zrange_in_sorted_set("sorted_set_zrange_operation")

3.执行测试

打开终端执行命令

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

参数解释

-u：并发用户数，配合--headless使用

-r：每秒生成用户的速率

-t：执行时间

执行结果