locust_

简介：

代码仓库：https://github.com/locustio/locust

官网：https://locust.io/

官方文档：https://docs.locust.io/en/stable/

论坛：https://stackoverflow.com/questions/tagged/locust?tab=newest&page=2&pagesize=15&newreg=dbb6de1fe20940f7a9b4b24984994562

压测的基本原则：

1.施压策略 —-持续缓慢增压（找出最大用户数）

2.要考虑施压的环境：施压的环境（网络，设备性能），服务端

自建一个本地服务，通过修改施压脚本找到最优选择

服务端：提供五个接口
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
from flask import Flask, request, jsonify
import  requests,time,json,re
from lxml import etree

'''
----------------------------------------------
 启动命令： 
----------------------------------------------  
    linux: 
    需要安装： pip install Flask gunicorn
    gunicorn -w 4 -b 0.0.0.0:8000 app:app   
    第一个app ：  文件名
    第二app  为文件中app对象
    
    window： 
    waitress-serve --listen=*:8000 app:app
    需要安装：      pip install Flask waitress
    第一个app ：  文件名
    第二app  为文件中app对象
----------------------------------------------  
一共5个接口（3个get接口，2个post）
GET接口：
  测试 ip:port
  速度快： ip:port/get/test
  速度慢： ip:port/get
POST接口：
  速度快： ip:port/post/test
  速度慢： ip:port/post
----------------------------------------------
注意：
    除了测试的get接口， 其余需要在headers里加token参数
    post请求需要添加请求体
----------------------------------------------
请求头demo:

md5 = hashlib.md5()
md5.update('demo'.encode('utf-8'))
headers={
   'token': md5.hexdigest(),
   'Content-Type':'application/json'
}
----------------------------------------------
请求体demo：

data={
      't':int(time.time())
   }
----------------------------------------------
'''



app = Flask(__name__)

def get_ansor():
    res=requests.get('https://www.shiyebian.net/jiangsu/nanjing/').text
    HTML=etree.HTML(res)
    ul_list=HTML.xpath('//div[@class="main bg-white"]//ul')
    data = []
    for li in ul_list:
        urls=li.xpath('./li/a/@href')
        titles=li.xpath('./li/a/text()')
        if len(urls) == len(titles):
            for i in range(len(urls)):
                data.append({'url': urls[i], 'title': titles[i].encode('latin-1').decode('GBK')})
        else:
            print({'code':3000,"message":'error'})
    return  data

def num_to_chinese_units(num):

    chinese_units = {'亿': 100000000, '万': 10000}
    result = []
    remainder = num
    for unit, value in chinese_units.items():
        if remainder >= value:
            quotient, remainder = divmod(remainder, value)
            result.append(f'{quotient:.2f}{unit}')
    if remainder != 0:
        result.append(f'{remainder:.2f}元')
    return ' '.join(result)

def get_skd():
    t = int(time.time() * 1000)
    url = f'https://push2.eastmoney.com/api/qt/clist/get?cb=jQuery112301302125560766778_{t}&pn=1&pz=500&po=1&np=1&fields=f12%2Cf13%2Cf14%2Cf62&fid=f62&fs=m%3A90%2Bt%3A2&ut=b2884a393a59ad64002292a3e90d46a5&_={t + 1}'
    res = requests.get(url,verify=False).text
    res_json = json.loads(re.findall(r'\((.*?)\)', res)[0])
    data=[]
    for i in range(0, 6):
        disk = {}
        disk['name'] = res_json['data']['diff'][i]['f14']
        disk['price'] = num_to_chinese_units(res_json['data']['diff'][i]['f62'])
        data.append(disk)
    return  data

def validate_token(token):
    if len(token) ==32:
        return  True
    else:
        return False

def token_required(f):
    def decorator(*args, **kwargs):
        token = request.headers.get('token')
        if not token:
            return jsonify({'message': 'Token is missing!'}), 401
        if not validate_token(token):
            return jsonify({'message': 'Invalid token!'}), 403
        return f(*args, **kwargs)
    decorator.__name__ = f.__name__
    return decorator


@app.route('/get', methods=['GET'])
@token_required
def get_():
    params=get_ansor()
    response_data = {
        'message': 'ok',

        'data':params
    }
    return jsonify(response_data), 200

@app.route('/get/test', methods=['GET'])
@token_required
def get_test():
    response_data = {
        'message': 'ok',
        'data':'done'
    }
    return jsonify(response_data), 200
@app.route('/post/test', methods=['POST'])
@token_required
def post_():
    print(request.data)
    if request.is_json:

        response_data = {
            'message': 'ok',
            'data': 'done'
        }
        return jsonify(response_data), 200
    else:
        return jsonify({'message': 'error'}), 400
@app.route('/post', methods=['POST'])
@token_required
def post_test():
    print(request.data)
    if request.is_json:
        data = get_skd()
        response_data = {
            'message': 'ok',
            'data': data
        }
        return jsonify(response_data), 200
    else:
        return jsonify({'message': 'error'}), 400

@app.route('/', methods=['GET'])
def all():
    return '这是locust 测试页'

if __name__ == '__main__':
    app.run()
施压脚本demo： (host ，脚本名称根据实际调整)
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
105
106
107
108
109
110
111
112
113
114
115
116
117
import uuid

from locust import HttpUser, TaskSet, task, between,FastHttpUser
import json,hashlib
import os,time
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class GetL(TaskSet):
    def on_start(self) :
        md5 = hashlib.md5()
        md5.update('sss'.encode('utf-8'))
        self.headers=  {
            'token': md5.hexdigest(),
            "Content-Type": "application/json",
        }

    @task
    def get_(self):
        get_url_2 = "/get"
        with self.client.get(get_url_2, headers=self.headers, catch_response=True, name='get请求',
                             verify=False) as get_response:
            if "ok" in get_response.text:
                get_response.success()
            else:
                get_response.failure('get请求报错')


class GetQ(TaskSet):
    def on_start(self):

        md5 = hashlib.md5()

        md5.update('sss'.encode('utf-8'))
        self.headers = {
            'token': md5.hexdigest(),
            "Content-Type": "application/json",
        }

    @task
    def get_Test(self):
        get_url = "/get/test"
        with self.client.get(get_url, headers=self.headers, catch_response=True, name='get_test请求',
                             verify=False) as get_response:

            if "done" in get_response.text:
                get_response.success()
            else:
                get_response.failure('get_test请求报错')

class PostQ(TaskSet):
    def on_start(self) :

        md5 = hashlib.md5()

        md5.update('sss'.encode('utf-8'))
        self.headers=  {
            'token': md5.hexdigest(),
            "Content-Type": "application/json",
        }

    @task
    def post_(self):
        post_url = '/post/test'
        data = {
            't': int(time.time())
        }
        with self.client.post(post_url, headers=self.headers, catch_response=True, json=data, name='post_test请求',
                              verify=False) as get_response:
            if "done" in get_response.text:
                get_response.success()
            else:
                get_response.failure('post_test请求报错')


class PostL(TaskSet):

    def on_start(self):

        md5 = hashlib.md5()

        md5.update('sss'.encode('utf-8'))
        self.headers = {
            'token': md5.hexdigest(),
            "Content-Type": "application/json",
        }

    @task
    def post_(self):
        post_url = '/post'
        data = {
            't': int(time.time())
        }
        with self.client.post(post_url, headers=self.headers, catch_response=True, json=data, name='post请求',
                              verify=False) as get_response:
            if "ok" in get_response.text:
                get_response.success()
            else:
                get_response.failure('post请求报错')

#使用 FastHttpUser相较于 HttpUser  更快
class WebsiteUser(FastHttpUser):
    # tasks = [PostQ,PostL,GetQ,GetL]  #随机选择
    #有权重的选择 ---数值越大请求的越多
    # tasks={
    #     PostQ:1,
    #     PostL:2,
    #     GetQ:1,
    #     GetL:2
    # }
    #若跑单个接口
    tasks=[PostQ]
    host = "http://10.1.1.114:3000"
    wait_time = between(1, 2)
if __name__ == '__main__':
    cmd = 'locust -f locust_day1.py'
    os.system(cmd)

locust2
locust_demo