Locust、JMeter性能测试之查询类接口【二】

本文为博主原创，未经授权，严禁转载及使用。
本文链接：https://blog.csdn.net/zyooooxie/article/details/124111115

前面分享 第1期 Locust、JMeter性能测试之查询类接口【一】，现在把第二期补上。

【实际这篇博客推迟发布N个月】

个人博客：https://blog.csdn.net/zyooooxie

【以下所有内容仅为个人项目经历，如有不同，纯属正常】

Locust-辅助工具

我使用的是 Locust 2.8.2

"""
@blog: https://blog.csdn.net/zyooooxie
"""

def read_txt_user_gen(file: str = r'D:\0002-十万用户_手机信息.txt', nrows: int = 10, skiprows: int = None, all_data: str = None):
    if skiprows is None:
        skiprows = random.randint(10, 90000)

    if all_data is not None:
        Log.info('读取TXT文件 全部数据')
        data = pd.read_csv(file, encoding='utf-8', dtype=object)

    else:
        Log.info('{}-{}'.format(nrows, skiprows))

        data = pd.read_csv(file, encoding='utf-8', skiprows=skiprows, nrows=nrows, dtype=object)

    all_data_gen = (d for d in data.values)

    return all_data_gen

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

"""
@blog: https://blog.csdn.net/zyooooxie
"""

import os
import math
import locust.stats
import queue
import traceback
import random

from locust import between, constant, constant_pacing, constant_throughput
from locust import FastHttpUser
from locust import HttpUser
from locust import task
from user_log import Log
from locust import LoadTestShape
from requests_toolbelt.utils import dump

from xx.xxxx import read_txt_user_gen

locust.stats.CONSOLE_STATS_INTERVAL_SEC = 30

txt_data = read_txt_user_gen(all_data='YES', file=r'D:\60w.txt')


# class QueryUserInfo(HttpUser):
class QueryUserInfo(FastHttpUser):

    # wait_time = between(1, 10)            # 在每次任务执行后引入延迟。如果未指定wait_time，则下一个任务将在完成后立即执行。
    wait_time = constant(1)

    host = 'https://blog.csdn.net/zyooooxie'

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.queue_obj = queue.Queue()

        self.user_id = None
        # Log.info('开工-{}--{}'.format(self, self.queue_obj))

    def first(self):

        self.queue_obj.put_nowait(next(txt_data)[0])

    @task
    def query(self):
        self.first()

        try:
            self.user_id = self.queue_obj.get_nowait()

        except queue.Empty:

            Log.info('Queue Empty:{}'.format(self))
            # exit()

        url = '/queryUserInfo'
        test_dict = {"userId": self.user_id, "qq": "153132336"}
        test_header = {'version': '1.0'}

        with self.client.post(url, json=test_dict, headers=test_header,
                              catch_response=True, name='queryUserInfo') as res:

            try:
                # # 更改断言，有些user在表里就是查不到
                # assert res.text.find(self.user_id) != -1

                assert res.json().get('success') is True
                res.success()

            # except AssertionError:
            #     Log.info(dump.dump_all(res).decode('utf-8'))
            #
            #     res.failure("AssertionError")

            except Exception as e:
                Log.debug(dump.dump_all(res).decode('utf-8'))

                Log.debug(traceback.format_exc())

                # Log.info(e.with_traceback(sys.exc_info()[2]))

                res.failure("{}".format(repr(e)))

            finally:
                # self.queue_obj.put_nowait(self.user_id)
                # Log.info('{}-{}'.format(self.user_id, self))

                pass
				
				
				
if __name__ == '__main__':

    pass

    os.system('del *.log')
	
    cmd_m = 'locust -f {} --logfile=locust.log '.format(__file__)
    Log.info(cmd_m)

    os.system(cmd_m)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

Locust’s web interface

https://docs.locust.io/en/stable/quickstart.html#locust-s-web-interface

without the web UI

https://docs.locust.io/en/stable/configuration.html

https://docs.locust.io/en/stable/running-without-web-ui.html

本地运行

locust -f D:/performance_test/queryUserInfo0317.py --csv=example  --csv-full-history  --headless -u 105 -r 15 --run-time 1m --stop-timeout 10  --html example_locust.html --logfile example_locust.log
1

https://docs.locust.io/en/stable/running-distributed.html#running-distributed

本机-分布式

通常应该在工作机器上的每个处理器内核上运行一个工作实例，以利用它们的所有计算能力。

locust -f D:/performance_test/queryUserInfo0317.py --csv=example  --csv-full-history  --headless -u 105 -r 10 --run-time 3m --stop-timeout 10 --expect-workers 2 --html example_locust.html --logfile example_locust.log  --master


locust -f D:/performance_test/queryUserInfo0317.py --worker

1
2
3
4
5

压测结果

本文链接：https://blog.csdn.net/zyooooxie/article/details/124111115

个人博客 https://blog.csdn.net/zyooooxie