Locust性能压测工具-支持分布式压测【多测师】

ccbpm火线安全 lv.1

发布时间：2022-11-13 15:35:47 329

相关标签： # less# python# ssh# 工具

为什么用locust做压测？？
1.因为locust可以完美兼容python
2.locust采用的是协程，LoadRunner 和 Jmeter 这类采用进程和线程的测试工具，都很难在单机上模拟出较高的并发压力。
Locust 的并发机制摒弃了进程和线程，采用协程（gevent）的机制。协程避免了系统级资源调度，由此可以大幅提高单机的并发能力
3.设置集合点需要自己写代码控制
4.并发数量多的情况下要用分布式，locust完美支持分布式还有无界面形式设置master主节点和slave子节点。
5.使用Locust进行性能测试时，当一台单机不足以模拟所需的用户数量的时候，可以在多台机器上分布式的执行性能测试
缺点：没有像jmeter可以设置并发数量的概念，locust只要启动会一直不断的发送请求。

代码：
#coding=utf-8
import requests
from locust import HttpLocust,TaskSet,task,seq_task
import os
from locust import events
from gevent._semaphore import Semaphore
all_locusts_spawned = Semaphore()
all_locusts_spawned.acquire()
def on_hatch_complete(**kwargs):
    all_locusts_spawned.release()  #创建钩子方法

events.hatch_complete += on_hatch_complete    #挂载到locust钩子函数（所有的Locust实例产生完成时触发）

class Cms(TaskSet):
    # 访问cms后台首页
    @seq_task(1)
    def cms_login(self):
        # 定义请求头
        header = {
            "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.67 Safari/537.36"
}

        req = self.client.get("/cms/manage/loginJump.do?userAccount=admin&loginPwd=123456",  headers=header, verify=False)
        if req.status_code == 200:
            print("success")
        else:
            print("fails")

    @seq_task(2)
    def cms_queryUserList(self):
        header = {
            "User-Agent":"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.67 Safari/537.36"
}
        queryUserList_data = {
            "startCreateDate": "",
            "endCreateDate": "",
            "searchValue": "",
            "page": "1"
}
        req = self.client.post("/cms/manage/queryUserList.do",data=queryUserList_data)
        queryUserList_url = 'http://192.168.23.132:8080/cms/manage/queryUserList.do'
if req.status_code == 200:  #u'查询用户成功！')
            print("success")
        else:
            print("fails")

    def on_start(self):
        """ on_start is called when a Locust start before any task is scheduled """
        # self.cms_login()
        # self.cms_queryUserList()
        all_locusts_spawned.wait()    #限制在所有用户准备完成前处于等待状态

class websitUser(HttpLocust):
    task_set = Cms
    min_wait = 1000  # 单位为毫秒
    max_wait = 1000  # 单位为毫秒

if __name__ == "__main__":
    import os
    os.system("locust -f lesson8.py --host=http://192.168.23.134:8080")


'''
semaphore是一个内置的计数器：
每当调用acquire()时，内置计数器-1
每当调用release()时，内置计数器+1
计数器不能小于0，当计数器为0时，acquire()将阻塞线程直到其他线程调用release()
'''

其他参考资料介绍：

http://www.testclass.net/locust --locust详细介绍
https://testerhome.com/topics/16766https://testerhome.com/topics/19000 -- 自动化测试平台
https://www.missshi.cn/api/view/blog/59a6b53fe519f50d04000103 --locust性能测试工具深度剖析
https://testerhome.com/topics/17068 --wrk ab jmeter locust性能压测工具对比
https://blog.debugtalk.com/categories/3-Testing/性能测试/ --tearshome社区

文章来源： https://blog.51cto.com/u_15846941/5821555

特别声明：以上内容（图片及文字）均为互联网收集或者用户上传发布，本站仅提供信息存储服务！如有侵权或有涉及法律问题请联系我们。