locust之参数化

1. 在性能测试中，最基本就是多用户并发测试，而不同用户发起的请求，必然是会有不一样的，最简单的就是登录，每个用户的信息都会不一样。

jmeter或者LR最常见的就是使用参数化文件，关联变量，每个用户获取不同的值。

locust没有现成的组件可用，需要自己写代码去实现，简单的就是使用队列，在测试前把测试参数全部写入队列，每个用户执行前从队列中取值。

如果测试完所有参数就终止，则队列空时结束。

如果需要循环测试，则取出值后，再把值写回队列，则队列永远也不会空了。

 

2.队列queue介绍

Queue

Queue.Queue(maxsize=0)

Queue提供了一个基本的FIFO（先进先出）容器，使用方法很简单,maxsize是个整数，指明了队列中能存放的数据个数的上限。一旦达到上限，插入会导致阻塞，直到队列中的数据被消费掉。如果maxsize小于或者等于0，队列大小没有限制。

 

Queue创建的实例对象常用方法

queue.qsize()：返回当前队列包含的消息数量

queue.empty()：如果队列为空，返回True，反之False

queue.full()：如果队列满了，返回True，反之False

queue.put(item,[block[,timeout]]：将item消息写入队列，block默认值为True

1）bolck使用默认值（True）

①没有设置timeout（单位秒），消息队列如果已经没有空间可写入，此时程序将被阻塞（停在写入状态），直到从消息队列腾出空间为止

②设置了timeout，则会等待timeout秒，若还没空间，则抛出“queue.full”异常

2）block值为False，消息队列如果没有空间可写入，则会立刻抛出“queue.Full”异常

queue.put_nowait(item)：相当于queue.put(item,False)

queue.get([nlock[,timeout]])：获取队列中的一条消息，然后将其从队列中移除，block默认为True

1）bolck使用默认值（True）

①没有设置timeout（单位秒），消息队列如果为空，此时程序将被阻塞（停在读取状态），直到从消息队列读到信息为止

②设置了timeout，则会等待timeout秒，若还没读取到任何消息，则抛出“queue.Empty”异常

2）block值为False，消息队列如果为空，则会立刻抛出“queue.Empty”异常

queue.get_nowait()：相当于queue.get(False)
 

3.模板代码

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import queue
import time
import gevent
from locust import TaskSet, task, HttpUser, between, events


class MyTaskSet(TaskSet):
    """ 定义用户行为 """

    wait_time = between(1, 3)

    def on_start(self):
        print("Executing on_start ...")

    def on_stop(self):
        print("Executing on_stop ...")

    @task
    def login(self):
        """ 用户登录 """
        username = None
        try:
            username = self.user.queue_data.get()
            print(f"username: {username}")
            self.user.queue_data.put_nowait(username)
        except Exception as e:
            print(e)
            print("Queue is empty.")
        if not username:
            exit(1)
        jsondata = {"parameters": {"mobile": username, "captcha": "0000"}}
        path = "/api/account/phonesigninorregister"
        with self.client.post(path, json=jsondata, catch_response=True) as response:
            if response.status_code != 200:
                response.failure("Response code wrong!")
            try:
                rsp = response.json()
                print(rsp)
                if rsp["isSucceed"]:
                    response.success()
                else:
                    response.failure("Response is not success!")
                if rsp["statusCode"] != 200:
                    response.failure("Response code is not 200!")
            except Exception as e:
                print(e)
                response.failure("Response content wrong!")


class MyTeskGroup(HttpUser):
    """ 定义线程组 """
    tasks = [MyTaskSet]
    host = "http://192.168.2.129"
    # 初始化参数化队列
    queue_data = queue.Queue()
    queue_data.put_nowait("13900010000")
    queue_data.put_nowait("13900020000")
    queue_data.put_nowait("13900030000")
    queue_data.put_nowait("13900040000")
    queue_data.put_nowait("13900050000")
    queue_data.put_nowait("13900060000")
    queue_data.put_nowait("13900070000")
    queue_data.put_nowait("13900080000")
    queue_data.put_nowait("13900090000")
    queue_data.put_nowait("13900100000")