重学Locust

吴和硕

2023-12-01

一、Locust对比Jmeter

Locust官网：Locust - A modern load testing framework

Jmeter官网：Apache JMeter - Apache JMeter™

Locust和Jmeter经常会拿来做对比，在使用和学习Locust之前，我们来简单看下两个工具的对比，其实各优利弊。

参考：阿里性能专家全方位对比Jmeter和Locust，到底谁更香？ - 知乎

基本总结：

发压能力：相同并发下，Locust（使用FastHttpLocust）> Jmeter。
并发能力：Locust和Jmeter旗鼓相当，都能满足工作需求，Jmeter消耗的内存更高。
结果报表：Jmeter好于Locust，但是基本都满足工作需求。
学习成本：Jmeter>Locust。
易用性：Jmeter > Locust。

使用建议：

如果只是做简单的接口测试、压力测试，没有需要写代码来扩展的特殊需求，首选Jmeter。
如果某些测试场景需要写代码来扩展，你会Java的话，可以选择Jmeter。
如果某些测试场景需要写代码来扩展，你会Python的话，可以选择Locust。
如果想在单台机器发起更大的压力的话，并且Python代码能力不错的话，可以选择Locust，记得一定要使用FastHttpLocust客户端。

Locust官网曾提到过，默认情况下，Locust使用requests库发送HTTP请求，性能不太好，如果要产生更高的压力，建议使用FastHttpLocust作为HTTP客户端来压测，性能可以提升5-6倍。但是FastHttpLocust并不能完全替代requests库。

参考：Increase performance with a faster HTTP client — Locust 2.8.3 documentation

二、Locust的高级能力

Locust只内置了对HTTP/HTTPS的支持，但它可以扩展到测试几乎任何系统，比如：gRPC、Thrift、WebSocket、Kafka、Selenium/WebDriver等。参考：Testing non-HTTP systems — Locust 2.8.3 documentation

1、示例：编写gRPC协议

注意需要在打开通道之前执行以下代码，从而使gRPC gevent兼容：

import grpc.experimental.gevent as grpc_gevent

grpc_gevent.init_gevent()

被压服务：Server代码示例

import hello_pb2_grpc
import hello_pb2
import grpc
from concurrent import futures
import logging
import time

logger = logging.getLogger(__name__)


class HelloServiceServicer(hello_pb2_grpc.HelloServiceServicer):
    def SayHello(self, request, context):
        name = request.name
        time.sleep(1)
        return hello_pb2.HelloResponse(message=f"Hello from Locust, {name}!")


def start_server():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    hello_pb2_grpc.add_HelloServiceServicer_to_server(HelloServiceServicer(), server)
    server.add_insecure_port("localhost:50051")
    server.start()
    logger.info("gRPC server started")
    server.wait_for_termination()

发压端：压测代码示例

# make sure you use grpc version 1.39.0 or later,
# because of https://github.com/grpc/grpc/issues/15880 that affected earlier versions
import grpc
import hello_pb2_grpc
import hello_pb2
from locust import events, User, task
from locust.exception import LocustError
from locust.user.task import LOCUST_STATE_STOPPING
from hello_server import start_server
import gevent
import time

# patch grpc so that it uses gevent instead of asyncio
import grpc.experimental.gevent as grpc_gevent

grpc_gevent.init_gevent()


@events.init.add_listener
def run_grpc_server(environment, **_kwargs):
    # Start the dummy server. This is not something you would do in a real test.
    gevent.spawn(start_server)


class GrpcClient:
    def __init__(self, environment, stub):
        self.env = environment
        self._stub_class = stub.__class__
        self._stub = stub

    def __getattr__(self, name):
        func = self._stub_class.__getattribute__(self._stub, name)

        def wrapper(*args, **kwargs):
            request_meta = {
                "request_type": "grpc",
                "name": name,
                "start_time": time.time(),
                "response_length": 0,
                "exception": None,
                "context": None,
                "response": None,
            }
            start_perf_counter = time.perf_counter()
            try:
                request_meta["response"] = func(*args, **kwargs)
                request_meta["response_length"] = len(request_meta["response"].message)
            except grpc.RpcError as e:
                request_meta["exception"] = e
            request_meta["response_time"] = (time.perf_counter() - start_perf_counter) * 1000
            self.env.events.request.fire(**request_meta)
            return request_meta["response"]

        return wrapper


class GrpcUser(User):
    abstract = True

    stub_class = None

    def __init__(self, environment):
        super().__init__(environment)
        for attr_value, attr_name in ((self.host, "host"), (self.stub_class, "stub_class")):
            if attr_value is None:
                raise LocustError(f"You must specify the {attr_name}.")
        self._channel = grpc.insecure_channel(self.host)
        self._channel_closed = False
        stub = self.stub_class(self._channel)
        self.client = GrpcClient(environment, stub)


class HelloGrpcUser(GrpcUser):
    host = "localhost:50051"
    stub_class = hello_pb2_grpc.HelloServiceStub

    @task
    def sayHello(self):
        if not self._channel_closed:
            self.client.SayHello(hello_pb2.HelloRequest(name="Test"))
        time.sleep(1)

2、Using Locust as a library

可以从自己的Python代码启动负载测试，而不是使用蝗虫命令运行蝗虫。

通过启动Locust的Environment实例来完成，完整示例如下，具体内容及使用方法可参考：Using Locust as a library — Locust 2.8.3 documentation

import gevent
from locust import HttpUser, task, between
from locust.env import Environment
from locust.stats import stats_printer, stats_history
from locust.log import setup_logging

setup_logging("INFO", None)


class User(HttpUser):
    wait_time = between(1, 3)
    host = "https://docs.locust.io"

    @task
    def my_task(self):
        self.client.get("/")

    @task
    def task_404(self):
        self.client.get("/non-existing-path")


# setup Environment and Runner
env = Environment(user_classes=[User])
env.create_local_runner()

# start a WebUI instance
env.create_web_ui("127.0.0.1", 8089)

# start a greenlet that periodically outputs the current stats
gevent.spawn(stats_printer(env.stats))

# start a greenlet that save current stats to history
gevent.spawn(stats_history, env.runner)

# start the test
env.runner.start(1, spawn_rate=10)

# in 60 seconds stop the runner
gevent.spawn_later(60, lambda: env.runner.quit())

# wait for the greenlets
env.runner.greenlet.join()

# stop the web server for good measures
env.web_ui.stop()

3、丰富的插件

Locust原生支持的功能比较简单，没有做的很重，但是Github上有很多优秀的插件拓展，比如：存储结果及可视化展示、命令行工具参数等。

参考：GitHub - SvenskaSpel/locust-plugins: A set of useful plugins/extensions for Locust

示例：使用Timescale + Grafana持久化存储和展示Locust的压测结果数据

具体方法及说明参考：locust-plugins/locust_plugins/timescale at master · SvenskaSpel/locust-plugins · GitHub

三、Locust开源或官方示例

Locust官方文档：Installation — Locust 2.12.1 documentation

1、Locust官方示例

完整的代码地址：locust/examples at master · locustio/locust · GitHub（examples目录），下面给出常见的几个示例：

基本HTTP示例：https://github.com/locustio/locust/blob/master/examples/basic.py
gRPC示例：locust/examples/grpc at master · locustio/locust · GitHub
使用lib库灵活自动控制压测过程示例：https://github.com/locustio/locust/blob/master/examples/use_as_lib.py
使用效率更高的HTTP库（默认为python-requests库）：https://github.com/locustio/locust/blob/master/examples/fast_http_locust.py
- 详细解释说明见：Increase performance with a faster HTTP client — Locust 2.12.1 documentation

2、locust-plugins开源示例

完整代码地址：GitHub - SvenskaSpel/locust-plugins: A set of useful plugins/extensions for Locust（examples目录）

这个开源的代码库给出了WebSocket/SocketIO, Kafka, Selenium/WebDriver等的压测代码示例，大家可以参考。

RESTful风格的HTTP示例：locust-plugins/rest_ex.py at master · SvenskaSpel/locust-plugins · GitHub
socket简单示例：locust-plugins/socketio_ex.py at master · SvenskaSpel/locust-plugins · GitHub

四、Locust使用问题记录

1、用例级别的setup和teardown，以及host等参数的提取

Locust.setup、Locust.teardown在1.0及以上版本去除，分别由@events.test_start.add_listener、@events.test_stop.add_listener替代，具体的可以参考下面的代码示例或官网。
locust命令行参数的获取可以使用argparse获取，也可以使用locust自身的environment参数获取，具体的可以参考下面的代码示例或官网。

代码示例：locust_baidu_demo.py

运行命令：locust -f locust_demo.py -H "http://www.baidu.com/"

# -*- coding: UTF-8 -*-
"""
# rs
# 测试访问百度首页
"""
import time
import random
import argparse
from logzero import logger
from locust import HttpUser, TaskSet, task, events, HttpLocust


# 用例级别的setup：开始时执行一次
@events.test_start.add_listener
def on_test_start(environment, **kw):
    logger.info("test is starting")
    
    # 方法1: 使用environment获取host
    logger.info("environment: {}".format(environment))
    logger.info("environment host: {}".format(environment.host))

    # 方法2：使用argparse获取host参数
    parser = argparse.ArgumentParser()
    parser.add_argument('-H', '--host')
    args, unknown = parser.parse_known_args()

    logger.info("argparse host: {}".format(str(args.host)))


# 用例级别的teardown：结束时执行一次
@events.test_stop.add_listener
def on_test_stop(**kw):
    print("test is stopping")


class WebsiteTasks(TaskSet):
    """
    # 示例
    """
    
    def on_start(self):
        """
        # 压测任务启动
        :return:
        """
        # 请求url
        self.url = "/"

    @task(1)
    def index(self):
        """
        # 任务
        """
        response = self.client.get(self.url)
        # logger.info("response: {}".format(response.text))


class LocustDemo(HttpUser):
    """
    # 压测百度首页示例
    """
    tasks = [WebsiteTasks]
    min_wait = 100   # 单位为毫秒
    max_wait = 600   # 单位为毫秒

    # QA测试环境：使用命令行传递参数
    # host = "http://www.baidu.com/"

2、控制 Locust 进程的退出

下面是一个示例，此代码可以进入 locustfile.py 或导入到 locustfile 中的任何其他文件中）

如果满足以下任一条件，则将退出代码设置为非零：

* 超过 1% 的请求失败

* 平均响应时间超过 200 ms

* 响应时间的第 95 百分位大于 800 ms

import logging
from locust import events

@events.quitting.add_listener
def _(environment, **kw):
    if environment.stats.total.fail_ratio > 0.01:
        logging.error("Test failed due to failure ratio > 1%")
        environment.process_exit_code = 1
    elif environment.stats.total.avg_response_time > 200:
        logging.error("Test failed due to average response time ratio > 200 ms")
        environment.process_exit_code = 1
    elif environment.stats.total.get_response_time_percentile(0.95) > 800:
        logging.error("Test failed due to 95th percentile response time > 800 ms")
        environment.process_exit_code = 1
    else:
        environment.process_exit_code = 0