Locust性能测试实战

原创

晴空SunnySky

修改于 2023-10-10 09:32:39

6282

修改于 2023-10-10 09:32:39

文章被收录于专栏：QA在路上

一：性能测试理论知识

1.1：性能测试及其目的

性能测试的定义：

通过自动化的测试工具模拟多种正常、峰值以及异常负载条件来对系统的各项性能指标进行测试。负载测试和压力测试都属于性能测试，两者可以结合进行。

性能测试的手段：

是通过模拟真实业务从而向服务器发送大量并发请求进而对被测系统产生负载，分析被测系统在不同压力下的表现。

我们进行性能测试的常见目的如下：

a:评估系统的性能（在局域网测试环境或生产环境下，通过测试结果的分析评估当前系统的服务级别）。

b: 定位性能瓶颈（通过性能测试找出影响系统整体性能的关键步骤或过程，为系统调优提供方向性依据）。

c:验证调优结果(通过比对优化后和优化前的测试结果，确认性能优化策略是否生效)。

1.2性能测试的种类细分

1.2.1压力测试

通过逐步增加系统负载，测试系统性能的变化，并最终确定在什么负载条件下系统性能处于失效状态来获得系统能提供的最大服务级别的测试。

压力测试是逐步增加负载，使系统某些资源达到临界点。

1.2.2负载测试

通过逐步增加系统负载，测试系统性能的变化，并最终确定在满足性能指标的前提下，系统所能够承受的最大负载量的测试。

1.2.3稳定性测试

通过给系统加载一定的业务压力（如CPU资源在70%～90%的使用率）的情况下，运行一段时间，检查系统是否稳定。因为运行时间较长，所以通常可以测试出系统是否有内存泄露等问题。

1.2.4:容量测试

在一定的软、硬件条件下，在数据库中构造不同数量级的记录数量，通过运行一种或多种业务场景，在一定虚拟用户数量的情况下，获取不同数量级别的性能指标，从而得到数据库能够处理的最大会话能力、最大容量等。

1.2.5配置测试

通过对被测试软件的软硬件配置的测试。配置测试能充分利用有限的软硬件资源，发挥系统的最佳处理能力，同时可以将其与其他性能测试类型联合应用，为系统调优提供参考。

1.3性能测试的实施流程

在实施性能测试的过程中，整体工作流程是1:分析性能测试需求->2:设计性能测试方案->3:开发性能测试脚本->4：搭建性能测试环境->5：执行测试->6：分析结果后多轮测试进行验证优化->7：编写性能测试报告->8：编写性能测试总结报告

二：压测工具选型

我们团队选用Jmeter和Locust两种压测工具。

今天只针对Locust讲解。

2.1 Locust是什么？

locust官网地址：https://www.locust.io/

2.2 Locust中压测脚本开发

# -*- coding = utf-8 -*-
 
 from locust import HttpUser, TaskSet, task, constant, between
 import random
 import re
 import json
 import hashlib
 from assertpy import assert_that
 import time
 from faker import Faker
 
 fake = Faker(locale='zh_CN')
 
 # 生成随机日期，格式为yyyy-mm-dd
 random_date = fake.date(pattern="%Y-%m-%d", end_datetime=None)
 # unix时间格式的当前毫秒
 current_mis_time = int(round(time.time() * 1000))
 
 app_header = {'Host': 'pt2dingtalkf.zjk.taeapp.com',
 'Accept': 'application/json, text/plain, */*',
 'Origin': 'https://pt2dingtalkf.zjk.taeapp.com',
 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.90 Safari/537.36',
 'sign': 'd44e27d775d6160b9f3961e1561d1ab58de3b1617263fecb249e7cc88e6af3be',
 'Content-Type': 'application/json;charset=UTF-8',
 'Referer': 'https://pt2dingtalkf.zjk.taeapp.com/pro/index.html?corpid=dinga93c84f4d89688bf35c2f4657eb6378f&appid=2576&dd_nav_bgcolor=ffff943e',
 'Accept-Encoding': 'gzip, deflate',
 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
 'Connection': 'keep-alive'}
 
 
 app_token = '193bfbec50530ff9932f6f8479b14388c95f666782f9b6d1920cf230bf08bb48'
 app_data_save_url = 'https://pt2dingtalkf.zjk.taeapp.com/pro/v1/form/data/add'
 
 
 class UserTasks(TaskSet):
 # 替换请求报文中的部分数据
 def process_request_data(self, init_data, expected_to_be_replace, need_replace_to_data):
 page_pattern = re.compile(expected_to_be_replace)
 
 matchers = page_pattern.findall(str(init_data))
 for matcher in matchers:
 init_data = str(init_data).replace(str(matcher), str(need_replace_to_data))
 return init_data
 
 def process_headers(self, init_headers, request_parameters, token):
 parameters = str(str(request_parameters) + str(token)).encode('utf-8')
 sign_code = hashlib.sha256(parameters).hexdigest()
 init_headers['sign'] = sign_code
 return init_headers
 
 # App后台添加客户
 def add_save_customer_on_app(self):
 init_param = '请求参数'
 request_param = self.process_request_data(init_param, '压测客户', str(fake.address()))
 request_header = self.process_headers(app_header, request_param, app_token)
 add_customer_result = self.client.post(url=app_data_save_url, data=request_param.encode('utf-8'),
 headers=request_header).text
 # 返回新建客户的id
 return json.loads(add_customer_result)['result']['dataId']
 
 
 # App后台添加合同订单
 @task
 def add_save_order_on_app(self):
 init_param = ''
 request_param = self.process_request_data(init_param, '压测合同订单', str(fake.address()))
 # 替换新建合同时关联的客户id
 customer_id = self.add_save_customer_on_app()
 final_param = self.process_request_data(request_param, str(325388), str(customer_id))
 request_header = self.process_headers(app_header, final_param, app_token)
 add_order_result = self.client.post(url=app_data_save_url, data=final_param.encode('utf-8'),
 headers=request_header).text
 
 assert_that(str(add_order_result)).contains("操作成功")
 
 
 class WebsiteUser(HttpUser):
 wait_time = between(1, 3)
 tasks = [UserTasks]

2.3 使用Locust模拟浪涌负载

class StagesShape(LoadTestShape):
 stages = [
 {"duration": 60, "users": 1, "spawn_rate": 0.5},
 {"duration": 120, "users": 2, "spawn_rate": 0.5},
 {"duration": 240, "users": 3, "spawn_rate": 0.5},
 {"duration": 300, "users": 0, "spawn_rate": 1},
 ]
 
 def tick(self):
 run_time = self.get_run_time()
 
 for stage in self.stages:
 if run_time < stage["duration"]:
 tick_data = (stage["users"], stage["spawn_rate"])
 return tick_data
 
 return None

2.4 使用更高效的http-client

from locust.contrib.fasthttp import FastHttpUser
# App后台添加客户
 @task
 def add_save_customer_on_app(self):
 init_param = '请求参数'
 request_param = self.process_request_data(init_param, '压测客户', str(fake.address()))
 request_header = self.process_headers(app_header, request_param, app_token)
 add_customer_result = self.client.post(path=app_data_save_url, data=request_param.encode('utf-8'),
 headers=request_header).text
 
 assert_that(str(add_customer_result)).contains("操作成功")
class WebsiteUser(FastHttpUser):
 wait_time = between(1, 3)
 tasks = [UserTasks]

2.5 分布式并行执行

Locust的分布式并行执行使用master-worker模式：
启动master角色的进程：
locust -f stage_load.py --master --host=https://pt2dingtalkf.zjk.taeapp.com
启动worker角色的进程：
locust -f stage_load.py --worker --master-host=127.0.0.1

三：完成实例(架构升级)

3.1：实施背景

架构升级+子表单改造项目过中对底层作了很大改动，需验证架构升级后系统稳定性。

3.2：测试策略

分别在测试2(架构升级项目的代码)和测试3(master分支代码)环境上做同样业务场景的压测，对比压测结果；预期目标：架构升级后的各压测业务的性能指标不会差于升级前的性能。

3.3：压测场景设计

生产环境接口请求量top30收集(App+Web)

一：移动端

1.1 单据详情接口URL：/mobile/form/data/get 压测业务：客户，合同，PaaS表单。

1.2 单据基本信息接口URL：/mobile/form/data/basic/detail 压测业务：客户，合同，PaaS表单。

1.3 图表中心统计结果查看接口URL：/chart/result 压测业务：客户，合同，跟进记录。

1.4 列表接口URL：/mobile/list/customer 压测业务：客户，合同，PaaS表单。

1.5 新建保存接口URL：/mobile/form/data/add 压测业务：客户，合同，PaaS表单。

1.6 新建页 /mobile/form/data/add/get 压测业务：客户，合同，PaaS表单。

1.7 关联数据列表接口URL：/mobile/detail/tab/linkAdd/get 压测业务：子表单中的关联数据列表。

二：Web后台

1.8 Web单据新建页 /pro/v1/form/data/get 压测业务：客户，合同，PaaS表单。

1.9 Web单据详情接口URL：/pro/v1/form/data/detail 压测业务：客户，合同，PaaS表单。

1.10 Web列表页接口URL：/pro/v1/list/customer 压测业务：客户，合同，PaaS表单。

1.11 Web统计指标接口URL：/pro/v1/chart/result/value 压测业务：客户，合同图表查询。

3.4：压测脚本开发

由于压测场景的脚本较多，不一一列举。

4.1 Web后台新建保存客户

# Web后台添加客户
@task
def add_save_customer_on_web(self):
    init_param = '请求参数'
    request_param = self.process_request_data(init_param, '压测客户', str(fake.address()))
    request_header = self.process_headers(web_login_header, request_param, web_token)
    add_customer_result = self.client.post(url=web_data_detail_url, 		   data=request_param.encode('utf-8'),headers=request_header).text

    assert_that(str(add_customer_result)).contains("操作成功")

4.2 Web后台合同列表页翻页查看

# Web后台合同列表
@task
def order_list_on_web(self):
    init_param = '{\"corpid\":\"dinga93c84f4d89688bf35c2f4657eb6378f\",\"userId\":\"030917160122954929\",\"platform\":\"web\",\"sortMap\":{},\"formId\":3497,\"saasMark\":1,\"businessType\":201,\"subBusinessType\":201,\"timeFilter\":{},\"listGroupId\":0,\"defaultGroup\":1,\"commonFilter\":{},\"del\":0,\"page\":1,\"pageSize\":20,\"conditions\":[],\"statusFilter\":2,\"appId\":64,\"isSync\":true}'
    # 替换页数
    final_param = self.process_request_data(init_param, '\"page\":1',
                                            str('\"page\":' + str(random.randrange(1, 229))))
    request_header = self.process_headers(web_login_header, final_param, web_token)
    order_list_result = self.client.post(url="https://pt2pcf.zjk.taeapp.com/pro/v1/list/contract", data=final_param.encode('utf-8'),
                                           headers=request_header).text

    assert_that(str(order_list_result)).contains("操作成功")

4.3 Web后台查看PaaS表单数据的详情

# Web后台查看PaaS表单数据的详情
@task
def paas_detail_on_web(self):
    init_param = '{\"corpid\":\"dinga93c84f4d89688bf35c2f4657eb6378f\",\"userId\":\"030917160122954929\",\"platform\":\"web\",\"menuId\":\"3503\",\"appId\":110,\"dataId\":5125,\"saasMark\":2,\"businessType\":0,\"subBusinessType\":404}'
    # 替换表单id
    final_param = self.process_request_data(init_param, str(5125), str(random.randrange(4066, 5125)))

    request_header = self.process_headers(web_login_header, final_param, web_token)
    add_paas_result = self.client.post(url=web_data_detail_url, data=final_param.encode('utf-8'),
                                           headers=request_header).text

    assert_that(str(add_paas_result)).contains("操作成功")