FastHttpUser vs HttpUser, создание кастомного клиента для не-HTTP протоколов, request_meta.
Locust не ограничен HTTP. Любой протокол — WebSocket, gRPC, TCP, AMQP — можно протестировать, написав собственный клиент и правильно передавая метрики в статистику Locust.
FastHttpUser — это более быстрая альтернатива стандартному HttpUser. Он использует geventhttpclient вместо requests, что даёт меньший overhead CPU на каждый запрос.
Разница существенна при большом числе пользователей: HttpUser с 2000 пользователями может нагрузить CPU генератора до предела, тогда как FastHttpUser справляется с тем же числом при значительно меньшей нагрузке:
from locust import task, between
from locust.contrib.fasthttp import FastHttpUser
class FastApiUser(FastHttpUser):
wait_time = between(0.5, 1.5)
@task
def browse(self):
self.client.get("/api/items")
@task
def create(self):
self.client.post("/api/items", json={"name": "test"})API FastHttpUser практически идентично HttpUser: те же методы (get, post, put, delete), те же параметры name, catch_response=True. Основное ограничение: не поддерживает cert и некоторые расширенные параметры requests.
Стандартный HttpUser основан на requests.Session и хорошо подходит для сценариев, где нужна совместимость с экосистемой requests (middleware, custom adapters, сложная cookie-логика).
FastHttpUser использует geventhttpclient.HTTPClient — низкоуровневый клиент с минимальным overhead. Используйте FastHttpUser когда:
Для тестирования WebSocket-сервисов нужно создать собственный клиент и вручную записывать метрики через request_meta:
import time
import gevent
from locust import User, task, between, events
from websocket import create_connection
class WebSocketClient:
def __init__(self, host, environment):
self.host = host
self.environment = environment
self.ws = create_connection(f"ws://{host}/ws")
def send_and_receive(self, message, name="websocket"):
start_time = time.time()
exception = None
response_length = 0
try:
self.ws.send(message)
response = self.ws.recv()
response_length = len(response)
except Exception as e:
exception = e
finally:
elapsed_ms = int((time.time() - start_time) * 1000)
self.environment.events.request.fire(
request_type="WebSocket",
name=name,
response_time=elapsed_ms,
response_length=response_length,
exception=exception,
)
return response
def close(self):
self.ws.close()
class WebSocketUser(User):
wait_time = between(1, 3)
def on_start(self):
self.ws_client = WebSocketClient(self.host, self.environment)
def on_stop(self):
self.ws_client.close()
@task
def send_message(self):
self.ws_client.send_and_receive(
'{"type": "ping"}',
name="ws/ping"
)Ключевой момент — вызов self.environment.events.request.fire(). Именно через него метрики попадают в статистику Locust и отображаются в Web UI.
Метод request.fire() принимает именованные аргументы:
request_type — произвольная строка для типа запроса ("GET", "WebSocket", "gRPC")name — имя/путь запроса, под которым он появится в статистикеresponse_time — время в миллисекундахresponse_length — размер ответа в байтах (0, если неизвестен)exception — объект исключения если запрос упал, иначе NoneПри exception=None запрос считается успешным. При exception=SomeException(...) — неуспешным, с этим исключением в колонке Failures.
gRPC-сервисы тестируются аналогично — через базовый класс User (без Http) и вызов events.request.fire():
import time
import grpc
from locust import User, task, between
# Импорт сгенерированных protobuf-классов
from myservice_pb2 import HelloRequest
from myservice_pb2_grpc import GreeterStub
class GrpcUser(User):
wait_time = between(0.5, 2)
abstract = True # не создавать пользователей этого базового класса напрямую
def on_start(self):
self.channel = grpc.insecure_channel(f"{self.host}:50051")
self.stub = GreeterStub(self.channel)
def on_stop(self):
self.channel.close()
def grpc_call(self, stub_method, request, name):
start = time.time()
exception = None
response_length = 0
try:
response = stub_method(request)
response_length = response.ByteSize()
except grpc.RpcError as e:
exception = e
self.environment.events.request.fire(
request_type="gRPC",
name=name,
response_time=int((time.time() - start) * 1000),
response_length=response_length,
exception=exception,
)
class HelloUser(GrpcUser):
@task
def say_hello(self):
self.grpc_call(
self.stub.SayHello,
HelloRequest(name="Locust"),
name="/helloworld.Greeter/SayHello"
)Атрибут abstract = True указывает Locust не создавать пользователей этого класса напрямую — он служит базовым классом для переиспользования.
При наследовании от User (а не HttpUser) атрибут self.client недоступен. Это осознанный выбор: User — минималистичный базовый класс без предположений о протоколе. Все соединения создаются вручную в on_start.
Это подходит для тестирования:
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.