locust 试用简记

使用

性能指标

spawn rate 每秒增加的用户数
number of total users to simulate 模拟的用户数量

HttpUser 提供http操作的方法

任务管理：

1. task
   用于任务规划设定及权重
   方法前加 @task(1)
2. SequentialTaskSet 任务类
   顺序执行任务on_start -> open_001 -> open_002

class TaskCase(SequentialTaskSet):
  def on_start(self):
    print("on start")
  @task
  def open_001(self):
    print("open 001")
  @task
  def open_002(self):
    print("open 002")

等待时间：

between用于控制延迟停顿，使用户执行每个任务后，随机等待XX秒
wait_time = between(5, 9)
name="XXX" 相当于为请求分组
on_start可用用来作为 fixture

configuration file

locustfile = ./locustfiles/asr_locust.py
host = http://127.0.0.1:8089/
csv=./report/locust
csv-full-history=true
html=./report/locustreport.html
logfile=./report/locustlog.txt

# 无头浏览器，需指定run-time和users
#headless = true
# 指定时间后停止
#run-time = 10m
#users = 100
#spawn-rate = 10

# worker 分布式
#master = true
#expect-workers = 5

# 指定tag执行
#tags=[test]

locust 源码阅读理解

HttpSession

本质上还是requests.Session
改成这样，locust 和 接口自动化测试 都可以调用。

#from requests import Session
from locust.clients import HttpSession
from utils.get_data import api_data
from locust import events


class MySession(HttpSession):
    proxies = None
    if api_data()["is_use_fiddler"]:
        proxies = {"http": "localhost:8888", "https": "localhost:8888"}
    def __init__(self,request_event=events.request,user="",*args, **kwargs):
        super().__init__(base_url="",request_event = request_event,user=user,*args, **kwargs)

    def request(self, method, url, proxies=proxies, verify=False, *args, **kwargs):
        global resp
        if url.startswith("/"):
            url = api_data()["api_data"]["url"] + url

        resp = super(MySession, self).request(
            method,
            url,
            proxies=proxies,
            verify=verify,
            timeout=(3, 180),
            *args,
            **kwargs
        )
        return resp

FastHttpSession

如果有压力测试需求，推荐使用这个...

#from requests import Session
from locust.clients import HttpSession
from locust.contrib.fasthttp import FastHttpSession
from utils.get_data import api_data
from locust import events


class MyFastSession(FastHttpSession):
    proxies = None
    if api_data()["is_use_fiddler"]:
        proxies = {"http": "localhost:8888", "https": "localhost:8888"}
    def __init__(self, environment,user,*args, **kwargs):
        super().__init__(base_url="",environment = environment,user=user,*args, **kwargs)

    def request(self, method, url, proxies=proxies, verify=False, *args, **kwargs):
        global resp
        if url.startswith("/"):
            url = api_data()["api_data"]["url"] + url

        resp = super(MyFastSession, self).request(
            method,
            url,
            proxies=proxies,
            verify=verify,
            timeout=(3, 180),
            *args,
            **kwargs
        )
        return resp

以上详细源码后面贴到github

贴一下试用结果

这里用的是高德地图API

from utils.http_request import MyFastSession
from locust.contrib.fasthttp import FastHttpUser

class HttpLocust(FastHttpUser):
    wait_time = between(5, 9)

    def __init__(self, environment, *args, **kwargs):
        super(HttpLocust, self).__init__(environment, *args, **kwargs)
        self.client = MyFastSession(environment, self)
        self.client.headers = {"Accept": "application/json"}

    def on_start(self):
        print("--------- task start ------------")

    def on_stop(self):
        print("---------- task stop ------------")

    @task
    def test_geocode(self):
        url = api_data()["geocode_geo"]["api"]+"address="+api_data()["geocode_geo"]["address"]+"&key="+api_data()['key']

        self.client.get(url)

问题

结合pytest,requests,websocket-client执行接口自动化测试，以下出现的问题不一定适用于所有场景。

1. RecursionError: maximum recursion depth exceeded

RecursionError: maximum recursion depth exceeded
/home/test/.local/share/virtualenvs/speech-evaluation-api-testing-_AH93AeQ/lib/python3.7/site-packages/locust/__init__.py:3: MonkeyPatchWarning: Monkey-patching ssl after ssl has already been imported may lead to errors, including RecursionError on Python 3.6. It may also silently lead to incorrect behaviour on Python 3.7. Please monkey-patch earlier. See https://github.com/gevent/gevent/issues/1016. Modules that had direct imports (NOT patched): ['urllib3.util (/home/test/.local/share/virtualenvs/speech-evaluation-api-testing-_AH93AeQ/lib/python3.7/site-packages/urllib3/util/__init__.py)', 'urllib3.util.ssl_ (/home/test/.local/share/virtualenvs/speech-evaluation-api-testing-_AH93AeQ/lib/python3.7/site-packages/urllib3/util/ssl_.py)'].
   monkey.patch_all()

根据提示得到的issue获得解决方法，在conftest.py最前面添加以下代码：

import gevent
from gevent import monkey
gevent.monkey.patch_all()

2. RPS 为0
   原因：休眠时间设置太长了... 设置了15-19s

3. 实际场景如何测试？
   比如 10个用户，2个在下单，2个在闲逛，其余的在付款。
   思路： 写一个taskset类，然后多个user,user分别指定task. 不确定对不对...

扩展

locust boomer 据说性能更好

参考

入门级教程 - 简单看看
chat
官方文档
阿里性能专家全方位对比Jmeter和Locust，到底谁更香？

上面的内容，主要还是一些知识点整理和个人的一些思考，权当参考，如有错误或者更好的建议，可以在评论区指正，不胜感激！