当前位置: 首页 > 工具软件 > Locust > 使用案例 >

重学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 · GitHubexamples目录),下面给出常见的几个示例:

2、locust-plugins开源示例

完整代码地址:GitHub - SvenskaSpel/locust-plugins: A set of useful plugins/extensions for Locustexamples目录

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

 

 四、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
 类似资料: