(共556篇)
全部分类

Locust后端压力测试
[ 未分类 ] 

准备好 python 环境

win,Mac,Linux 各自准备好自己的 python 环境, 本文以python3+版本为例

pip 工具相当于 NPM, 是 python 的包管理器, 一般会随着 python 一起安装

1
2
3
4
5
python -V # 查看python版本
pip -V # 查看pip版本

# 设置阿里云镜像作为pip的安装源
pip config set global.index-url https://mirrors.aliyun.com/pypi/simple/

准备项目目录

在本机中任意位置创建项目目录, 比如/opt/www/locust-demo

为项目准备虚拟环境

1
2
3
4
5
# 安装虚拟环境
pip install pipenv

# 为当前项目创建虚拟环境
pipenv --three # 使用本机的python3版本创建虚拟可执行库

安装 locust 包

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
# 必须进入当前项目的虚拟环境再安装所需要的包
cd /opt/www/locust-demo

# 进入虚拟环境
pipenv shell

# 安装locust主包
pip install locust

# 安装加密所需要的包
pip install pycryptodome

简单的 locust 测试案例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
class AppIndexTest(HttpUser):

    # 在所有的测试开始前执行一次
    def on_start(self):
        print('before all test')

    # 在所有的测试结束后执行一次
    def on_end(self):
        print('end all test)

    @task
    def index_hot(self):
        # 这里模拟请求的是前端首页路径
        self.client.request('get','/')

要注意的是只有使用了@task 修饰器的函数, 才会被当做测试函数

执行测试案例

1
2
3
4
5
# 一定要进入虚拟环境执行locust
pipenv shell

# 启动lcoust服务
locust -u 10 -r 10 -H https://b.admalltech.com --headless -t 10s

参数解释:

1
2
3
4
5
6
-u 10 预期要启动的用户数(用户数可以理解为并发数)
-r 10 每秒启动的用户数, 达到-u的数量后, 会一直保持最大数量持续请求
-H    设置接口前缀
--headless  以cli的模式执行测试, 如果没有这个参数, locust会启动一个web服务, 可以通过浏览器来控制启动参数,如果通过浏览器控制, 其他参数都可以不要
-t  测试的时间周期, 支持h  m  s等单位
-f  如果你的启动文件不是默认的locustfile.py, 可以通过该参数来指定启动文件

web关键字说明

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
Type    请求类型
Name    请求路径
requests    当前请求的数量
fails    当前请求失败的数量
Median    中间值,单位毫秒,一般服务器响应时间低于该值,而另一半高于该值
Average    所有请求的平均响应时间,毫秒
Min    请求的最小的服务器响应时间,毫秒
Max    请求的最大服务器响应时间,毫秒
Content Size    单个请求的大小,单位字节
reqs/sec    每秒钟请求的个数

测试DEMO(自有项目)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
# encode.py	
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
from base64 import b64encode
CLIENT_ID = '3061177457'
CLIENT_SEC = '596y6w51dqdu8qa9akh8ilw7hyg44jw5'

KEY = b'admalltect202106'
IV = b'admalltect202106'


def aes_encode(data):
    cipher = AES.new(KEY, AES.MODE_CBC, IV)
    ct_bytes = cipher.encrypt(pad(data.encode(), AES.block_size))
    ct = b64encode(ct_bytes).decode('utf-8')
    return ct
  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
# locustfile.py
from encode import aes_encode
import hashlib
import json
import os
import time
from locust import HttpUser, task

CLIENT_ID = '3061177457'
CLIENT_SEC = '596y6w51dqdu8qa9akh8ilw7hyg44jw5'

opset = {
    "pro": {
        "username": '18516081294',
        "password": '123456',
        "activeId": '1544926730572140544',
        'productId': '1540602257043755008'
    },

    "test": {
        "username": '13164158426',
        "password": '111111',
        "activeId": '1541356036307423232',
        'productId': '1534831336278331392'
    }
}


ops = opset['pro']

print(ops)


class AppIndexTest(HttpUser):
    token = ''

    # def on_start(self):
    #     res = self.sign('post', '/admall-user/api/login/business-login-password', {
    #         "username": ops['username'],
    #         "password": aes_encode(ops['password']),
    #         'infoType': 'business',
    #     })
    #     self.token = json.loads(res)['content']['token']

    def sign(self, method, url='/', body={}):
        # 获取最新的header字段
        headers = {
            'os': 'ios',
            'platform': 'B2B',
            'client_id': CLIENT_ID,
            'timestamp': str(int(round(time.time()*1000))),
            'noncestr': os.urandom(16).hex(),
            'sign': '',
            'content-type': 'application/json',
            'Authorization': 'Bearer ' + self.token
        }
        data = json.dumps(body, separators=(',', ':')) if body else '{}'
        if(method == 'get'):
            sign = ''.join((url, '#', headers['timestamp'], '#',
                           CLIENT_SEC, '#', headers['noncestr'], '##'))
        else:
            sign = ''.join(
                (url, '#', headers['timestamp'], '#', CLIENT_SEC, '#', headers['noncestr'], '#', data, '#'))
        # print('signOrigin: '+sign)

        md5 = hashlib.md5()
        md5.update(sign.encode('utf-8'))
        headers['sign'] = md5.hexdigest().upper()

        with self.client.request(method=method, url=url, data=data, headers=headers)as res:
            if(res.ok == 0):
                print('Error:', method, res.url, res.content.decode('utf8'))

            return res.content

    # @task
    # def index_hot_product(self):
    #     self.sign('post', '/b2b/v1.0.0/product/hot',
    #               {"pageNum": 1, "pageSize": 10, "cityCode": "410100", "recommendedConfigCode": "APPhome"})

    #     self.sign('post', '/b2b/v1.0.0/product/hot',
    #               {"lng": "113.809588", "lat": "34.742319", "pageSize": 10, "sort": "CHOICE_SHOP", "pageNum": 1, })

    # @task
    # def banner_list(self):
    #     self.sign('post', '/admall-user/api/ad/b2b-app/banner/list',
    #               {"platform": "app-s2b", "locationCode": "JCSYZJ", "cityCode": "410100"})
    #     self.sign('post', '/admall-user/api/ad/b2b-app/banner/list',
    #               {"platform": "app-s2b", "locationCode": "HOME", "cityCode": "410100"})
    #     self.sign('post', '/admall-user/api/ad/b2b-app/banner/list',
    #               {"platform": "app-s2b", "locationCode": "SYTC", "cityCode": "410100"})

    # @task
    # def has_coupon(self):
    #     self.sign('get', '/admall-user/api/coupon/check/user/new', {})

    # # 活动专题页
    # @task
    # def get_act_product(self):
    #     self.sign('get', '/common/discount/detail/'+ops['activeId'], {})

    # # 商品详情页
    # @task
    # def list_product(self):
    #     self.sign('post', '/b2b/v1.0.0/evaluate/list/product',
    #               {"status": "1", "productId": ops['productId'], "pageSize": "1", "pageNo": 1})

    @task
    def product_detail(self):
        self.sign('get', '/b2b/v1.0.0/product/detail/' + ops['productId'], {})

    # @task
    # def footmark_save(self):
    #     self.sign('post', '/admall-user/api/b2b/footmark/save',
    #               {"productId":  ops['productId']})

    # @task
    # def user_hierarchy(self):
    #     self.sign('post', '/admall-basics/api/record/user/hierarchy',
    #               {"code": "B_BUYER_REG_VIEW_NOT_CONVERSION"})

    # @task
    # def supplier_query(self):
    #     self.sign('post', '/admall-user/api/message/center/app/supplier/query', {})

    # @task
    # def draw_check(self):
    #     self.sign('get', '/admall-user/api/coupon/product/draw/check/' + ops['productId'],
    #               {"token": ""})

    # @task
    # def product_detail_hot(self):
    #     self.sign('post', '/b2b/v1.0.0/product/hot',
    #               {"platform": "app-s2b", "cityCode": "410100", "pageSize": 10, "productId":  ops['productId'], "pageNum": 1})