Locust 压力测试工具

简单学习使用一个压力测试巩固

2017-03-06 | 阅读

使用Locust

Locust是一个基于Python的支持分别式的压力测试工具。其特点是,可以产生大量的用户,且用户行为由Python代码控制,且整个测试过程中有一个Web的UI界面可以实时观察测试数据。

安装

使用 pip install locustio安装 locustio

但是pip没有安装成功,前往https://pip.readthedocs.io/en/stable/installing/#install-pip 下载pip并安装。

在osx 中,需要手动安装libeventbrew install libevent.

快速使用

使用locust -h查看帮助,我们编写一个测试,需要编写一个locustfile.py文件,如下面的简单实例:

from locust import HttpLocust, TaskSet, task

class UserBehavior(TaskSet):
    def on_start(self):
        """ on_start is called when a Locust start before any task is scheduled """
        self.login()

    def login(self):
        self.client.post("/login", {"username":"ellen_key", "password":"education"})

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

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

class WebsiteUser(HttpLocust):
    task_set = UserBehavior
    min_wait = 5000
    max_wait = 9000

在上述代码中,我们定义了一堆任务,这些任务是一个python的会滴啊函数,并传递一个参数l,l为一个Locust类的实例。在类UserBehavior中,我们定义了用户行为,即在开始时,调用login函数,而后在tasks任务列表中,选择任务执行。

在最后的用户设置中, min_wait表示两个任务之间的最小间隔,而max_wait表示任务的超时时间。

编写好配置文件后,我们在该文件目录下,就可以执行locust命令来进行测试 :

locust --host=http://example.com

然后进入测试页面:

测试页面有两个内容需要填写,一个是用户数量,一个是每秒新增用户数量。用户数量是最大的用户数量,而新增用户数是每秒增加的用户数量。如果设置了每秒增加用户数量,则每秒会自动创建新的任务,直到达到最大值。

locustfile 的编写

HttpLocust这个类提供了Http请求的功能。使用这个类时,每一个实例都会获得一个属于HttpSession类型的实例client

from locust import HttpLocust, TaskSet, task

class MyTaskSet(TaskSet):
    @task(2)
    def index(self):
        self.client.get("/")

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

class MyLocust(HttpLocust):
    task_set = MyTaskSet
    min_wait = 5000
    max_wait = 15000

实例说明

from locust import HttpLocust, TaskSet, task

basePrePath = "xc"
imageName = "lenajpg.jpg?x-oss-process=image"
gmurl = "/resize,m_fixed,h_256,w_256/quality,q_50/format,webp"
# gmurl = ""
# 77831  4892

class UserBehavior(TaskSet):
    userCount = 0
    @task(1)
    def test(self):
        self.testtime = self.testtime + 1
        url = "/Testimage/%s%d%d/%s%s" % (basePrePath, self.index,self.testtime,imageName,gmurl)
        with self.client.request("GET",url, catch_response=True,name="256x256/50%/webp",headers={"host":"fffkkkkk.oss-cn-shanghai.aliyuncs.com"}) as response:
            # print response.headers['content-length']
            if response.headers['content-length'] == "4892":
                response.success()
            else:
                print "error gm url: " + url + "  contentLength = " + response.headers['content-length']
                response.failure("wrong response picture")
        # self.interrupt()


    def on_start(self):
        UserBehavior.userCount += 1
        self.index = UserBehavior.userCount 
        self.testtime = 0

class WebsiteUser(HttpLocust):
    task_set = UserBehavior
    min_wait = 10
    max_wait = 3000
    host = 'http://139.196.67.70'

我们以以上代码,简要说明一下一些需要注意的地方,首先任务组合是TaskSet派生类UserBehavior,我们通过这个类来定义任务以决定每个用户的动作。其中通过@task(1)来注解任务,其中值1表示其在任务集合中所占的比重。

TaskSet 拥有属性client,这个属性是HttpLocust的实例,而我们通过这个属性来发送请求。 在这次请求中,我们通过self.client.request方法,发起请求 ,

headers={"host":"fffkkkkk.oss-cn-shanghai.aliyuncs.com"}

表示设置一个header中的属性。

name="256x256/50%/webp"

表示对于可变类型的URL,我们赋值其一个固定的Tag,以便于统计。

catch_response=True 表示手动控制请求的成功与失败,一般会如下进行控制 :

with client.get("/does_not_exist/", catch_response=True) as response:
    if response.status_code == 404:
        response.success()

TaskSet中,通过on_start函数,来在该对象初始化时,进行一些处理。

WebsiteUser继承与HttpLocust,定义整个测试任务的一些属性,task_set表示当前测试任务集合,也就是上面的UserBehavior类。而min_waitmax_wait表示一个任务的最短和最长等待时间。