首页 > 其他分享> 文章详细

项目压测数据

2022-08-20 20:30:09 阅读：168 来源： 互联网

标签：name temp 项目 self token dict user 测数据

压测流程

首先启动 locust 压测脚本
然后启动bus查分模拟脚本
收集数据
压测结束，清理数据

采集的数据为：

请求相关数据，如响应时间，请求总数据量
资源相关，请求时pod的数量以及实时cpu，内存消耗
请求数量数量，总请求数量，时间分布
apm请求记录，查询请求具体耗时
数据库信息，记录网络连接数变化
locust请求端数据总计，错误数量，RPS，耗时统计等

单测查询接口

单独测试查询结果的接口，要求：返回时间不超过1s，如果超过1s则添加pod，测试极限压力

查分实际场景模拟
在查分时有查询接口和添加准考证，而且还会有查分bus。查实多个接口调用下的各项参数指标。

数据准备
用户：cettest20~cettest79 共60个
获取用户token，保存在redis中，后续请求需要携带token发起请求。
注意：该接口只有staging环境才能使用

import requests
import redis

pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)


token_dict = {}
url = "https://apiv3.shanbay.com/bayuser/login"
for i in range(20,80):
    user_name = f"cettest{str(i)}"
    
    payload = {
        "account": user_name,
        "code_2fa":"",
        "password": "123456"
    }
    headers = {"X-API-TOKEN": "IBILiaNGtorE"}
    res = requests.post(url, json=payload, headers=headers)
    if 200 == res.status_code:
        print(user_name)
        token_dict[user_name] = res.cookies.get("auth_token")
    r.hmset("cettest_user_token_dict", token_dict)


print(token_dict)
token_dict_redis = r.hgetall("cettest_user_token_dict")
print(token_dict_redis)

locust 压测脚本

locust_test.py

import time
import random
import redis
from locust import HttpUser, TaskSet, task, constant_throughput


class UserBehavior(HttpUser):
    wait_time = constant_throughput(2)

    def on_start(self):
        pool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
        r = redis.Redis(connection_pool=pool)

        # cettest20~cettest79 共60个测试账号的token
        self.token_dict = r.hgetall("cettest_user_token_dict")  # 字典 {user_name: token ...}
        self.token_list = [token for token in self.token_dict.values()]
        self.user_list = []
        for i in range(20, 80):
            temp = {}
            temp["name"] = f"cettest{str(i)}"
            temp["ticket"] = str(random.randint(5,9))*15
            temp["phone_num"] = 11111111111
            temp["source_plan_id"] = random.randint(19,20)
            self.user_list.append(temp)

    @task(4)
    def get_user_ticket(self):
        url = "https://apiv3.shanbay.com/tickethub/ticket/user?code=2021-12-cet"
        cookies = {"auth_token": random.choice(self.token_list)}
        self.client.get(url, cookies=cookies)
    
    @task(2)
    def post_user_ticket(self):
        url = "https://apiv3.shanbay.com/tickethub/ticket/user?code=2021-12-cet&is_in_shanbay=1"
        user = random.choice(self.user_list)
        payload = {
            "channel": 0,
            "name": user["name"],
            "phone_num": user["phone_num"],
            "source_plan_id": user["source_plan_id"],
            "ticket": user["ticket"]
        }
        cookies = {"auth_token": self.token_dict[user["name"]]}
        self.client.post(url, cookies=cookies, json=payload)

    @task(1)
    def put_user_ticket(self):
        url = "https://apiv3.shanbay.com/tickethub/ticket/user?code=2021-12-cet&is_in_shanbay=1"
        user = random.choice(self.user_list)
        payload = {
            "name": user["name"],
            "phone_num": user["phone_num"],
            "ticket": user["ticket"]
        }
        cookies = {"auth_token": self.token_dict[user["name"]]}
        self.client.put(url, cookies=cookies, json=payload)

启动脚本：

locust -f  locust_test.py

访问：
127.0.0.1:8098 填写用户数和每秒增长数
RPS计算：代码中的 wait_time 是一次请求的间隔时间，为1表示请求一次间隔1s，RPS为1。用户数指开启的协程数。所以RPS = wait_time * 用户数
bus 查分模拟脚本
过程：查询出post请求创建的用户，调用查分结果的bus写入分数。
数据特征：所有测试用户的手机号都是 11111111111 学校 college为 "压测学院"

import time
import random
from app import models as am
from app.buses import send_user_grades


user_tickets = am.UserTicket.select().where(am.UserTicket.phone_num=="bQ8AqNMnFh8pLkOX5tXg9A==")

while True: 
    for user in user_tickets:
        print(f"{user.id}{user.name}")
        user_list = []
        temp = dict()
        temp["name"] = user.name
        temp["ticket"] = user.ticket
        temp["status"] = 0
        temp["reading"] = random.randint(100,300)
        temp["listening"] = random.randint(100,300)
        temp["writing"] = random.randint(100,300)
        temp["total"] = random.randint(300,500)
        temp["school_name"] = "压测学院"
        temp["exam_level"] = random.randint(0,1)
        user_list.append(temp)
        send_user_grades.delay(user_list)
    time.sleep(5)
    print("*" * 20)

数据清除脚本
根据数据特征清除测试数据

from app import models as am
user_tickets = am.UserTicket.select().where(am.UserTicket.phone_num=="bQ8AqNMnFh8pLkOX5tXg9A==", am.UserTicket.college=="压测学院").count()

print(f"将要删除的用户数{user_tickets}")
delete_users = am.UserTicket.delete().where(am.UserTicket.phone_num=="bQ8AqNMnFh8pLkOX5tXg9A==", am.UserTicket.college=="压测学院").execute()
print("删除的用户{delete_users}")

基准测试

PRS 10

测试时长：3 min
locust

grafana 请求数量

grafana pod资源

kibana 请求数量

kibana apm

rps 680

grafana 请求相关

grafana 资源相关

kibana 请求数量相关

标签：name,temp,项目,self,token,dict,user,测数据
来源： https://www.cnblogs.com/goldsunshine/p/16608521.html