个人中心
文章发布
退出
作者热门文章
- Java锁的逻辑（结合对象头和ObjectMonitor）
- 还在用饼状图？来瞧瞧这些炫酷的百分比可视化新图形（附代码实现）⛵
- 自动注册实体类到EntityFrameworkCore上下文，并适配ABP及ABPVNext
- 基于Sklearn机器学习代码实战
32
4
Locust基于python的协程机制，打破了线程进程的限制，可以能够在一台测试机上跑高并发 。
1.快慢：衡量系统的处理效率：响应时间 。
2.多少：衡量系统的处理能力：单位时间内能处理多少个事务（tps） 。
性能测试根据测试需求最常见的分为下面三类 。
1 负载测试load testing 。
不断向服务器加压，值得预定的指标或者部分系统资源达到瓶颈，目的是找到系统最大负载的能力 。
2 压力测试 。
通过高负载持续长时间，来验证系统是否稳定 。
3 并发测试:
同时像服务器提交请求，目的发现系统是否存在事务冲突或者锁升级的现象 。
性能负载模型 。
安装存在问题，可以通过豆瓣源下载 。
pip install locust
基本上多数的场景我们都可以基于这个模板read.py去做修改 。
from
locust
import
HttpUser, TaskSet, task, tag, events
#
启动locust时运行
@events.test_start.add_listener
def
setup(environment, **
kwargs):
#
print("task setup")
#
停止locust时运行
@events.test_stop.add_listener
def
teardown(environment, **
kwargs):
print
(
"
task teardown
"
)
class
UserBehavor(TaskSet):
#
虚拟用户启用task运行
def
on_start(self):
print
(
"
start
"
)
locusts_spawned.wait()
#
虚拟用户结束task运行
def
on_stop(self):
print
(
"
stop
"
)
@tag(
'
test1
'
)
@task(
2
)
def
index(self):
self.client.get(
'
/yetangjian/p/17320268.html
'
)
@task(
1
)
def
info(self):
self.client.get(
"
/yetangjian/p/17253215.html
"
)
class
WebsiteUser(HttpUser):
def
setup(self):
print
(
"
locust setup
"
)
def
teardown(self):
print
(
"
locust teardown
"
)
host
=
"
https://www.cnblogs.com
"
task_set
=
task(UserBehavor)
min_wait
= 3000
max_wait
= 5000
注：这里我们给了一个webhost，这样我们可以直接在浏览器中打开locust 。
当然我们可以把集合点操作放入上述模板的setup中去运行起来 。
locusts_spawned =
Semaphore()
locusts_spawned.acquire()
def
on_hatch_complete(**
kwargs):
"""
select_task类的钩子函数
:param kwargs:
:return:
"""
locusts_spawned.release()
events.spawning_complete.add_listener(on_hatch_complete)
n
=
0
class
UserBehavor(TaskSet):
def
login(self):
global
n
n
+= 1
print
(f
"
第{n}个用户登陆
"
)
def
on_start(self):
self.login()
locusts_spawned.wait()
@task
def
test1(self):
#
catch_response获取返回
with self.client.get(
"
/yetangjian/p/17253215.html
"
,catch_response=
True):
print
(
"
查询结束
"
)
class
WebsiteUser(HttpUser):
host
=
"
https://www.cnblogs.com
"
task_set
=
task(UserBehavor)
wait_time
= between(1,3
)
if
__name__
==
'
__main__
'
:
os.system(
'
locust -f read.py --web-host="127.0.0.1"
'
)
在上面两个例子中我们已经看到了一些，例如装饰器 events.test_start.add_listener ； events.test_stop.add_listener用来在负载测试前后进行一些操作，又例如on_start、 on_stop，在task执行前后运行，又例如 task ，可以用来分配任务的权重 。
等待时间 。
#
wait between 3.0 and 10.5 seconds after each task
#
wait_time = between(3.0, 10.5)
#
固定时间等待
#
wait_time = constant(3)
#
确保每秒运行多少次
constant_throughput（task_runs_per_second）
#
确保每多少秒运行一次
constant_pacing(wait_time)
同样也可以在User类下发重写wait_time来达到自定义 。
tag标记 。
@tag(
'
test1
'
)
@task(
2
)
def
index(self):
self.client.get(
'
/yetangjian/p/17320268.html
'
)
通过对任务打标记，就可以在运行时候执行运行某一些任务:
#只执行标记test1
os.system('locust -f read.py --tags test1 --web-host="127.0.0.1"')
#不执行标记过的
os.system('locust -f read.py --exclude-tags --web-host="127.0.0.1"')
#除去test1执行所有
os.system('locust -f read.py --exclude-tags test1 --web-host="127.0.0.1"')
自定义失败 。
#
定义响应时间超过0.1就为失败
with self.client.get(
"
/yetangjian/p/17253215.html
"
, catch_response=
True) as response:
if
response.elapsed.total_seconds() > 0.1
:
response.failure(
"
Request took too long
"
)
#
定义响应码是200就为失败
with self.client.get(
"
/yetangjian/p/17320268.html
"
, catch_response=
True) as response:
if
response.status_code == 200
:
response.failure(
"
响应码200，但我定义为失败
"
)
自定义负载形状 。
自定义一个shape.py通过继承 LoadTestShape 并重写tick 。
这个形状类将以100块为单位,20速率的增加用户数，然后在10分钟后停止负载测试（从运行开始的第51秒开始user_count会round到100） 。
from
locust
import
LoadTestShape
class
MyCustomShape(LoadTestShape):
time_limit
= 600
spawn_rate
= 20
def
tick(self):
run_time
=
self.get_run_time()
if
run_time <
self.time_limit:
#
User count rounded to nearest hundred.
user_count = round(run_time, -2
)
return
(user_count, self.spawn_rate)
return
None
运行图如下所示 。
通过命令行去触发 。
os.system(
'
locust -f read.py,shape.py --web-host="127.0.0.1"
'
)
不同时间阶段的例子 。
from
locust
import
LoadTestShape
class
StagesShapeWithCustomUsers(LoadTestShape):
stages
=
[
{
"
duration
"
: 10,
"
users
"
: 10,
"
spawn_rate
"
: 10
},
{
"
duration
"
: 30,
"
users
"
: 50,
"
spawn_rate
"
: 10
},
{
"
duration
"
: 60,
"
users
"
: 100,
"
spawn_rate
"
: 10
},
{
"
duration
"
: 120,
"
users
"
: 100,
"
spawn_rate
"
: 10
}]
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
。
最后此篇关于python轻量级性能工具-Locust的文章就讲到这里了,如果你想了解更多关于python轻量级性能工具-Locust的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。
32
4
0
在我的场景中，我在没有 Web UI 的情况下运行 Locust。我正在使用的命令是 locust -f my_locust_file --no_web -c 20 -r 4 # as a h
我想使用 Locust.IO 运行峰值测试，其中大量请求与我的服务并行。 我已经尝试过 locust，这是我想运行的命令类型: locust -f locustfile.py --headless -
在 Locust 文档中，我们只能使用 self.interrupt() 停止任务，但它将移至父类。它不会停止负载测试。我想在所有用户登录并完成任务后停止完整的负载测试 蝗虫版本:1.1 class
我正在测试一个内部网站，在播放时我看到了这个错误: POST /Ex_ACO_EI/Details/2433: 'SSLError(MaxRetryError("HTTPSConnectionPool
我们正在尝试实现基于用户场景的负载测试，它基本上在一个 url 中创建一个订单，然后我们必须在下一个 url 中传递订单 ID 以获取该订单的状态。 我们正在使用蝗虫顺序任务。因为我们希望它像第一个请
是否可以将 headless 浏览器与 Locust 集成？我需要我的负载测试来处理在每次页面加载时触发额外请求的客户端脚本。 最佳答案 这是一个老问题，但我现在遇到了它，并在“realbrowser
是否可以将 headless 浏览器与 Locust 集成？我需要我的负载测试来处理在每次页面加载时触发额外请求的客户端脚本。 最佳答案 这是一个老问题，但我现在遇到了它，并在“realbrowser
我在运行 locust 的同一台机器上运行 htop。在我今天早上运行的测试中，我看到一个 CPU(共 4 个)达到 100%，而其他 CPU 基本上处于空闲状态。我还观察到多达 8 个 locust
我正在考虑将我的多线程 python 脚本移动到 locust。 我的脚本所做的简单解释是: 为每个用户创建一个话题 在每个线程中验证用户并获取auth cookie 使用该 auth cookie
我有一个用于维护多个 locust 脚本的存储库，以对我的许多目标主机/服务进行负载测试。 如何将这些脚本集成到 k8s 集群之一上的 stable/locust 的 Helm 安装中？ 我们目前在不
我正在尝试使用 locust 登录我的 Web 应用程序。我刚开始使用蝗虫。 我正在使用以下代码登录应用程序。 post_data = {'username': username, 'password
我使用 pip 在我的 ubuntu 上安装了 locustio "pip install locustio" 问题是每次我在控制台上运行任何 locust 命令时(例如) locust --vers
一.简介 locust官网介绍：Locust 是一个用于 HTTP 和其他协议的开源性能/负载测试工具。其对开发人员友好的方法允许您在常规 Python 代码中定义测试。Locust测试可以从命令行
一.简介 locust官网介绍：Locust 是一个用于 HTTP 和其他协议的开源性能/负载测试工具。其对开发人员友好的方法允许您在常规 Python 代码中定义测试。Locust测试可以从命令行
我尝试在 locustio/locust docker 镜像中运行 locustfile，但它找不到 locustfile，尽管 locustfile 存在于 locust 目录中。 ~ docker
我目前正在使用环境变量将自定义参数传递给我的负载测试。例如，我的测试类如下所示: from locust import HttpLocust, TaskSet, task import os clas
我无法理解 Locust 结果，因为这是第一次对我的服务器进行负载测试，我在本地时间 00:00 使用命令行运行了 Locust； 1000 个用户，每秒 100 个孵化器和 10000 个请求。下面
我想模拟两个用户同时运行不同的任务。如何使用 Locust 性能测试框架做到这一点？ 最佳答案 只需定义多个用户类: class User1(HttpUser): @task def
我无法理解 Locust 结果，因为这是第一次对我的服务器进行负载测试，我在本地时间 00:00 使用命令行运行了 Locust； 1000 个用户，每秒 100 个孵化器和 10000 个请求。下面
我想模拟两个用户同时运行不同的任务。如何使用 Locust 性能测试框架做到这一点？ 最佳答案 只需定义多个用户类: class User1(HttpUser): @task def
我是一名优秀的程序员,十分优秀！