1.安装环境
安装redis
brew install redis开启redis服务
brew services start redis停止redis服务
brew services stop redis安装Python库
pip install locust redis2.编写脚本
loadTest.py
# coding=utf-8
import json
import random
import time
import redis
from locust import User, events, task, tag
from typing import Optionalhost_name = "localhost"
port_no = 6379def _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 0events.request.fire(request_type=command,name=key,response_time=total_time,response_length=response_length,exception=None)return resultexcept 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 Noneclass 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：执行时间
执行结果
