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
