Locust，非常简单易用、分布式、python 开发的压力测试工具。有图形化界面，支持将压测数据导出.

官网：https://locust.io/

文档：https://docs.locust.io/

GitHub：https://github.com/locustio/locust

安装：

pip3 install locust

locust -V

测试步骤：

[1] - 编写Python用户脚本

[2] - 使用locust命令执行性能测试

[3] - (可选)通过Web界面监测结果

测试步骤示例

1. 编写压测脚本

test.py

from locust import HttpUser, task

class HelloWorldUser(HttpUser):
    @task
    def hello_world(self):
        self.client.get("/hello") #先访问 /hello
        self.client.get("/world") #再访问 /word

或

from locust import HttpUser, TaskSet, task

# 定义用户行为
class UserBehavior(TaskSet):

    @task
    def baidu_index(self):
        self.client.get("/")

class WebsiteUser(HttpUser):
    task = [UserBehavior] # 指向一个定义的用户行为类
    min_wait = 3000 # 执行事务之间用户等待时间的下界（单位：毫秒）
    max_wait = 6000 # 执行事务之间用户等待时间的上界（单位：毫秒）

2. 启动压测

locust -f  test.py --host=https://www.baidu.com

3. 压测监控

http://localhost:8089

Number of users to simulate 模拟用户数

Hatch rate (users spawned/second) 每秒钟增加用户数

点击 "Start swarming" 进入压测页面

压测界面右上角有:被压测的地址、当前状态、RPS、失败率、开始或重启按钮

性能测试参数

• Type 请求的类型，例如GET/POST
• Name 请求的路径
• Requests 当前请求的数量
• Fails 当前请求失败的数量
• Median 中间值，单位毫秒，请求响应时间的中间值
• Average 平均值，单位毫秒，请求的平均响应时间
• Min 请求的最小服务器响应时间，单位毫秒
• Max 请求的最大服务器响应时间，单位毫秒
• Average size 单个请求的大小，单位字节
• Current RPS 代表吞吐量(Requests Per Second的缩写)，指的是某个并发用户数下单位时间内处理的请求数。等效于QPS，其实可以看作同一个统计方式，只是叫法不同而已。

自定义压测代码

qps_test.py

#!/usr/bin/python3
#!--*-- coding: utf-8 --*--
import time
import threading
import requests

NQ = 1
COUNT = 100
api = "http://test.com/?picurl="

def qps_tc():
    time_start = time.time()
    req = requests.get(api)
    assert req.status_code == 200
    time_cost = time.time() - time_start
    # print('time cost', time_cost, 'sec', req.status_code)


if __name__ == "__main__":
    #single thread
    print('Single thread perform search action one by one', COUNT, 'times', 'nq=', NQ)
    time_start = time.time()
    for k in range(COUNT):
        qps_tc()
    time_cost = time.time() - time_start
    print('Totally time cost', time_cost, 'sec')
    print('QPS', COUNT/time_cost)
    print('\n=================================================\n')

    #multi thread
    print('Multi-thread perform search action parallelly', COUNT, 'times', 'nq=', NQ)
    threads = []
    time_start = time.time()
    for k in range(COUNT):
        x = threading.Thread(target=qps_tc, args=())
        threads.append(x)
        x.start()
    for th in threads:
        th.join()
    time_cost = time.time() - time_start
    print('Totally time cost', time_cost, 'sec')
    print('QPS', COUNT / time_cost)

go 压测工具

• go实现的压测工具【单台机器100w连接压测实战】