Locust performance test - parameterization, no repetition of concurrent cyclic data sampling

background

Performance testing , Sometimes someone API Requests can only be processed once for the same data ( Such as user registration ), Or can only be executed sequentially , Multiple users are not allowed to operate on the data at the same time , Different data can be requested concurrently . Therefore, data sampling needs to be handled , Avoid multiple users sampling the same data at the same time .

Python queue

queue yes Python Standard library in , Commonly known as queue , Can directly import quote .

stay Python in , Data between multiple threads is shared , When multiple threads exchange data , Can't guarantee the security and consistency of data , So when multiple threads need to exchange data , The queue appears , Queues can perfectly solve the data exchange between threads , Ensure the security and consistency of data between threads .

The queue will pass the first in first out or first in last out mode , It ensures that a single data will not be accessed by multiple threads at the same time .

queue Modules have three types of queues and constructors :

If maxsize Less than 1 That means the queue length is infinite

FIFO  First in first out queue . class queue.Queue(maxsize=0)

LIFO  It's like a heap , First in, then out . class queue.LifoQueue(maxsize=0)

 The lower the priority queue level, the first to come out . class queue.PriorityQueue(maxsize=0)

queue Common methods in modules :

queue.qsize()  Returns the size of the queue 
queue.empty()  If the queue is empty , return True, conversely False
queue.full()  If the queue is full , return True, conversely False
queue.full  And  maxsize  Size correspondence 
queue.get([block[, timeout]]) Get the queue ,timeout Waiting time 
queue.get_nowait()  Quite a queue.get(False)
Queue.put(item, [block[, timeout]])  Written to the queue ,timeout Waiting time 
queue.put_nowait(item)  Quite a queue.put(item, False)
queue.task_done()  After finishing a job ,queue.task_done() Function sends a signal to the queue that the task has completed 
queue.join()  It essentially means waiting until the queue is empty , Do something else 

Locust application

stay HttpUser Class , Generate queue data

class TestLocust(HttpUser):

    queue_list = queue.Queue()
    for i in range(1, 10):
        queue_list.put_nowait(i)
 

stay task In the method, we use self.parent.queue_list.get() To visit queue Data in . If you want to cycle the queue data , You can put the extracted data back queue in , self.parent.queue_list.put_nowait(data)

class UserBehavior(TaskSet):

    @task  
    def get_root(self):
        data = self.parent.queue_list.get()
        print("queue data:{}".format(data))
        # self.parent.queue_list.put_nowait(data)

The following is a practical example , Let's see Locust Multi user concurrency queue Application of queues . In order to better show the concurrent process , Add some log Information , Every task Interval between 5s, simulation 2 individual user Concurrent .

1. Cyclic data retrieval , The data is not repeated

from locust import TaskSet, task, HttpUser
import os
from locust.user.wait_time import between
import queue

class UserBehavior(TaskSet):

    def on_start(self):
        print('taskset start')
        self.root_index = 0

    def on_stop(self):
        print('taskset end')

    @task  
    def get_root(self):
        print("get_root task")
        print("root index : " + str(self.root_index))
        self.root_index += 1
        
        if not self.parent.queue_list.empty():
            data = self.parent.queue_list.get()
            print("queue data:{}".format(data))
            response = self.client.get('',name='get_root')
        else:
            print("no data exist")
            exit(0)
            
        if not response.ok:
            print(response.text)
            response.failure('Got wrong response')

class TestLocust(HttpUser):
    # If no wait_time is specified, the next task will be executed as soon as one finishes.
    wait_time = between(5,5)

    def on_start(self):
        print('locust user start')

    def on_stop(self):
        print('locust user stop')
         
    tasks = [UserBehavior]
    host = "https://cn.bing.com"  

    queue_list = queue.Queue()
    for i in range(1, 6):
        queue_list.put_nowait(i) 

if __name__ == "__main__":
    # -u concurrency user number
    # -r generate user number per second 
    # --run-time or -t 
    os.system("locust -f test_locust.py --headless -u 2 -r 1 --run-time 20s --stop-timeout 5")

Output ：
It can be seen that 2 individual user All maintain their own variables root index, however queue The data of is taken out in the order of first in first out , until queue Empty , There's no duplicate data .

locust user start
taskset start
get_root task
root index : 0
queue data:1

locust user start
taskset start
get_root task
root index : 0
queue data:2

get_root task
root index : 1
queue data:3

get_root task
root index : 1
queue data:4

get_root task
root index : 2
queue data:5

get_root task
root index : 2
no data exist

2. Cyclic data retrieval , Duplicate data
take queue Put the retrieved data back queue A party , So the data in the series will not be empty .

from locust import TaskSet, task, HttpUser
import os
from locust.user.wait_time import between
import queue

class UserBehavior(TaskSet):

    def on_start(self):
        print('taskset start')
        self.root_index = 0

    def on_stop(self):
        print('taskset end')

    @task  
    def get_root(self):
        print("get_root task")
        print("root index : " + str(self.root_index))
        self.root_index += 1
        
        if not self.parent.queue_list.empty():
            data = self.parent.queue_list.get()
            print("queue data:{}".format(data))
            # put the data back to the queue
            self.parent.queue_list.put_nowait(data)
            response = self.client.get('',name='get_root')
        else:
            print("no data exist")
            exit(0)

        
        
        if not response.ok:
            print(response.text)
            response.failure('Got wrong response')

class TestLocust(HttpUser):
    # If no wait_time is specified, the next task will be executed as soon as one finishes.
    wait_time = between(5,5)

    def on_start(self):
        print('locust user start')

    def on_stop(self):
        print('locust user stop')
         
    tasks = [UserBehavior]
    host = "https://cn.bing.com"  

    queue_list = queue.Queue()
    for i in range(1, 6):
        queue_list.put_nowait(i) 

if __name__ == "__main__":
    # -u concurrency user number
    # -r generate user number per second 
    # --run-time or -t 
    os.system("locust -f test_locust.py --headless -u 2 -r 1 --run-time 20s --stop-timeout 5 --logfile log.txt --csv=example")

Output ：
It can be seen that 2 individual user All maintain their own variables root index, however queue The data of is taken out in the order of first in first out , And then put the fetched number back to the end of the queue , therefore queue Never empty , Cyclic data retrieval , The data will repeat .

locust user start
taskset start
get_root task
root index : 0
queue data:1

locust user start
taskset start 
get_root task 
root index : 0
queue data:2

get_root task
root index : 1
queue data:3

get_root task
root index : 1
queue data:4

get_root task
root index : 2
queue data:5

get_root task
root index : 2
queue data:1

get_root task
root index : 3
queue data:2

get_root task
root index : 3
queue data:3

taskset end
locust user stop
taskset end
locust user stop