locust的使用

1. 安装 locust

   pip install locust

2. 查看是否安装成功，查看版本号 locust -V
3. 更新最新版本号   pip install -U --pre locust

demo：

from locust import FastHttpUser, TaskSet, between, task, events, LoadTestShape
import os
from queue import Queue
from gevent._semaphore import Semaphore
 
all_locusts_spawned = Semaphore()
all_locusts_spawned.acquire()
n = 0
 
def on_hatch_complete(**kwargs):
    """Select_task类的钩子方法"""
    all_locusts_spawned.release()
 
 
events.spawning_complete.add_listener(on_hatch_complete)
 
 
class UserBehavior1(TaskSet):
    def login(self):
        global n
        n += 1
        # print("%s个虚拟用户开始启动，并登录"%n)
 
    def logout(self):
        print("退出登录")
 
    def on_start(self):
        self.login()
        self.id = self.parent.q.get()
        self.token = self.parent.test_data[self.id%4]
        # all_locusts_spawned.wait()   # 集合点
        print(f"{self.id}任务进入on_start")
 
    def on_stop(self):
        print(f"{self.id}任务退出on_stop")
        self.parent.q.put(self.id)
 
    @task
    def test_baidu_localnews(self):
        url = '/widget?id=LocalNews&ajax=json&t=1632650885696'
        param = {"limit": 2,"offset": 0,}
        # print(f"--------{self.token}------")
        with self.client.get(url,params=param,headers={},catch_response = True) as response:
            if response.json().get('errno') == 0:
                response.success()
            else:
                response.failure(f'{self.id}------Failed!')
 
 
class UserBehavior2(TaskSet):
 
    @task(5)
    def test_baidu_error(self):
        url = '/widget?id=China&ajax=json&t=1632650885696'
        param = {"limit": 2,"offset": 0,}
        with self.client.get(url, params=param, headers={}, catch_response=True) as response:
            if response.json().get('errno') == 0:
                response.success()
            else:
                response.failure(f'Failed!')
 
    @task(5)
    def test_baidu_ad(self):
        url = '/widget?id=ad&ajax=json'
        param = {"limit": 2, "offset": 0, }
        with self.client.get(url, params=param, headers={}, catch_response=True) as response:
            if response.json().get('errno') == 0:   #.status_code == 200:
                response.success()
            else:
                response.failure('Failed!')
 
 
# class MyCustomShape(LoadTestShape):  # 自定义并发数
#     stages = [
#         {"time": 10, "users": 10, "spawn_rate": 10},
#         {"time": 20, "users": 20, "spawn_rate": 10},
#         {"time": 30, "users": 10, "spawn_rate": 10},
#         {"time": 40, "users": 20, "spawn_rate": 10},
#     ]
#     def tick(self):
#         run_time = self.get_run_time()
#         for stage in self.stages:
#             if run_time < stage['time']:
#                 tick_data = (stage['users'],stage['spawn_rate'])
#                 return tick_data
#         return None
 
 
 
class WebsiteUser(FastHttpUser):
    q = Queue()
    for i in range(1, 1000):
        q.put(i)
    host = 'http://news.baidu.com'
    # tasks = {UserBehavior1: 1, UserBehavior2: 2}
    tasks = [UserBehavior1]
    test_data = ['1300000000', '1300000001', '1300000002', '1300000003']
 
    wait_time = between(0.9, 1.1)
 
if __name__ == '__main__':
 
    os.system("locust -f main.py")

脚本解析：

1、通过@task(n)装饰的方法为一个事务，方法的参数用于指定该行为的执行权重，参数越大每次被虚拟用户执行的概率越高，默认为1
2、 wait_time = between(1, 3)
来模拟用户的真实操作，例如用户在实际操作中，从上一步到下一步的操作，中间可能存在思考时间 ，between(1, 3) 表示用户思考的时间，随机为1到3秒

3、 host = 'http://news.baidu.com' 为要测试的域名

4、on_start ：每个虚拟用户在启动时都会调用该方法，获取token或测试用户数据等都放在这里

     on_stop ： 当虚拟用户用户停止运行（被终止）时调用

     每个虚拟用户只执行一次on_start和on_stop函数

5、HttpUser只被执行一次，因此很多公共的东西可以写到这里，所带属性在taskset中可以通过parent取出来。

6、 Semaphore常用于设置集合点（等待所有user都创建，再运行），用于对某些接口进行狭义的压力测试，以便进行调优。通常不用集合点，也就是这部分代码可以删除

7、检查点，请求函数需设定catch_response=True，会根据success failure进行统计

8、LoadTestShape可以自定义并发数量和时间，来模拟实际场景：时间峰值策略、时间阶段负载策略、逐步负载策略等等

 locust性能测试：（十）自定义负载策略 - 简书
 

参数解析：
 
Type：请求的类型，例如GET/POST。
Name：请求名称
requests：当前已完成的请求数量
fails：当前失败的数量
Median：响应时间的中间值，单位为毫秒
90%ile：根据正态分布，90%的响应时间在正态分布平均值下方
Average：平均响应时间，单位为毫秒
Min：最小响应时间，单位为毫秒
Max：最大响应时间，单位为毫秒
average Size：平均每个请求的数据量，单位为字节
current RPS（requests per second）：每秒钟处理请求的数量，即RPS

性能测试要考虑的问题：

1.网络带宽

2.线程池、数据库连接池（出现大量的等待链接的情况）

3.内存、CPU的利用率、Disk I/O、Network I/O

定位瓶颈：

1.压测流量是否进入后端：a.网络接入层由于带宽、最大连接数、新建连接数等限制     b.单ip地址限流     c. slb自动伸缩失败，nginx负载均衡失败或配置受限    d. 熔断、降级、限流

2.服务器端 硬件指标是否有问题    a.cpu高  top命令查看资源利用率  java应用可用jstack看出此线程正在执行的堆栈，看资源消耗在哪个方法上    b.内存高  看哪个进程占用大以及是否有大量的swap（虚拟内存交换）    c.磁盘 I/O高  减少日志输出、异步或换速度快的硬盘   d. 网络I/O  考虑网络传输内容的大小，不能超过硬件网络传输最大值的70%，可以通过压缩减少内容大小，在本地设置缓存以及分多次传输等操作提高网络I/O性能

3.中间件相关指标，线程池、连接池、JVM、kafka、redis、mq等

4.数据库相关指标，例如：慢查询SQL、命中率、锁、参数设置

5.JVM参数不合理，容量配置不合理、数据库设计不合理、程序架构规划不合理、程序本身设计有问题

调优：

1.把之前message服务提供缓存操作全部改为当前服务直连

2.优化所有服务的dubbo连接池 druid连接池 redis连接池，使之能够互相匹配，不会出现1000个dubbo链接等待一个数据库链接的现象

3.优化nginx 、gateway网关配置，宽带配置

4.使用sentinel哨兵来监控管理接口，可实时配置熔断等机制

5.使用arthas调试优化各方法内部耗时

6.修改各种同步机制改为异步操作

7.优化代码内部各种循环调用，重复请求，增加进程缓存

8.优化数据库sql，增加数据库索引            例如：max(id) 比order by id desc limit 1运行慢

9.广泛在使用缓存处采用双检查机制防止穿透数据库

压测报告：1.测试概述  目的、背景、指标和术语  2.测试概要  测试环境、人力资源、测试及性能指标收集工具  3.测试方案  数据构造规则、分布式压测准备、压测策略  4. 测试结果及结论 压测数据、压测结论   5.调优过程  定位瓶颈和调优过程  、性能测试过程中遇到的问题和解决方法  

通常的业务范围：1.关键业务  2.访问量大   3.逻辑复杂    4.运营推广活动  5.消耗资源

网络通常会设计大压力下的 熔断、 降级、 限流。因此JMeter中 TPS与线程数的关系，前段随着线程数增加，TPS也增加；中段线程数继续增加，TPS不变；后段随着线程数增加，TPS不变或下降

JMeter中常用的线程组：setup thread group、 Concurrency Thread Group  (阶梯式并发线程组 常用于探最大TPS) 、 Ultimate Thread Group   (最终线程组 用来模拟浪潮式的压测场景)、teardown thread group