Использование Gatling. Тестирование AMQP
Всем привет! С вами Сергей из команды тестирования производительности. Мы завершаем цикл статей о Gatling. Ранее мы уже рассмотрели работу с HTTP, JDBC, gRPC и Kafka, напоследок расскажу про AMQP.
Немного определений
Википедия дает такое определение протокола: Advanced Message Queuing Protocol — открытый протокол для передачи сообщений между компонентами системы. Основная идея в том, что отдельные подсистемы или независимые приложения могут обмениваться произвольным образом сообщениями через AMQP-брокер, который маршрутизирует, обеспечивает гарантированную доставку и распределяет потоки данных или подписку на нужные типы сообщений.
AMQP основан на трех понятиях:
Сообщение, message. Единица передаваемых данных, содержание которой никак не интерпретируется сервером. К сообщению могут быть присоединены структурированные заголовки.
Точка обмена, exchange. В нее отправляются сообщения. Точка обмена распределяет сообщения в одну или несколько очередей. При этом в точке обмена сообщения не хранятся.
Очередь, queue. Здесь хранятся сообщения до тех пор, пока их не заберет клиент. Он всегда забирает сообщения из одной или нескольких очередей.
Producer — клиентское приложение, которое публикует сообщения в exchange.
Consumer — клиентское приложение, которое получает сообщения из очереди сообщений.
Тестовый сервис RabbitMQ
Для разработки скрипта Gatling нам необходим сервер с AMQP-брокером. В этой статье используем RabbitMQ как наиболее распространенный. Чтобы развернуть тестовый сервис, необходим установленный docker. С помощью docker-compose поднимем сервис:
version: "3.9"
services:
rabbitmq:
image: rabbitmq:3.9.7-management-alpine
container_name: rabbitmq
environment:
- RABBITMQ_DEFAULT_USER=rabbitmq
- RABBITMQ_DEFAULT_PASS=rabbitmq
ports:
- '5672:5672'
- '15672:15672'
Запускаем:
docker-compose up
Теперь по адресу http://localhost:15672/ доступна консоль администратора. Логин и пароль для входа — rabbitmq.
Для скрипта публикации сообщений создаем очередь, в которую будем отправлять сообщения. Для этого нужно зайти в консоль, перейти на вкладку Queues, указать имя очереди test_queue и нажать AddQueue. Для скрипта публикации и чтения сообщений нужно по аналогии создать очередь test_queue_out.
Разработка скрипта публикации сообщений
Мы не будем разрабатывать проект с нуля, а используем уже готовый шаблон и с его помощью создадим проект myRabbitmq. Процесс создания мы описывали в первой статье цикла.
В результате получим готовый проект, но по умолчанию он для HTTP-протокола, поэтому его нужно переписать для AMQP.
Шаг 1. Обновим зависимости. В файле project/Dependencies.scala добавим (вместо <current version> подставить актуальную версию).
lazy val amqpPlugin: Seq[ModuleID] = Seq(
"ru.tinkoff" %% "gatling-amqp-plugin",
).map(_ % "<current version>" % Test)
В файле build.sbt добавим новую зависимость, чтобы после загрузки обновлений sbt можно было использовать AMQP-плагин.
libraryDependencies ++= amqpPlugin,
Загрузим новые зависимости в проект, запустив команду в консоли:
sbt update
Шаг 2. Введем переменные сервиса. В файле src/test/resources/simulation.conf хранятся дефолтные переменные для запуска. Добавим в него переменные, которые определяют наш RabbitMQ.
amqpHost: "localhost"
amqpLogin: "rabbitmq"
amqpPassword: "rabbitmq"
amqpPort: 5672
Шаг 3. Создадим Action для публикации сообщений. В директории сases создадим новый файл для объекта AmqpActions. В нем создадим действие, которое описывает публикацию сообщения в нашу очередь test_queue. Содержимое сообщения не статично и может изменяться в ходе теста. Параметр #{messageId} позволяет использовать значения, получаемые из Feeder.
package ru.tinkoff.load.myRabbitmq.cases
import io.gatling.core.Predef._
import ru.tinkoff.gatling.amqp.Predef._
import ru.tinkoff.gatling.amqp.request.{PublishDslBuilder, RequestReplyDslBuilder}
object AmqpActions {
val publishMessageToQueue: PublishDslBuilder = amqp("Publish to exchange").publish
.queueExchange("test_queue") // имя очереди, в которую отправляем сообщения
.textMessage("Hello message - #{messageId}") // текст сообщения
.messageId("#{messageId}") // настройки сообщения
.priority(0)
}
Шаг 4. Используем Feeders. Больше узнать о Feeders можно из вводной статьи про Gatling. В нашем примере используем фидеры из подключаемой библиотеки gatling-picatinny.
Возьмем базовый фидер Gatling, который читает данные из CSV-файла. Создадим в resources директорию pools и в ней файл messageIds.csv. Запишем в этот файл первой строкой заголовок messageId и далее в каждой строке значения от 1 до 10. В результате файл messageIds.csv должен выглядеть следующим образом, значения от 1 до 10:
messageId
1
2
…
10
В директории myRabbitmq создадим новую директорию feeders, а в ней object Feeders.
package ru.tinkoff.load.myRabbitmq.feeders
import io.gatling.core.Predef._
import io.gatling.core.feeder.BatchableFeederBuilder
object Feeders {
val idFeeder: BatchableFeederBuilder[String] = csv("pools/messageIds.csv").random
}
Шаг 5. Напишем сценарий теста. В CommonScenario описываем класс CommonScenario, в котором создаем сценарий — порядок выполнения определенных действий.
package ru.tinkoff.load.myRabbitmq.scenarios
import io.gatling.core.Predef._
import io.gatling.core.structure.ScenarioBuilder
import ru.tinkoff.gatling.amqp.Predef._
import ru.tinkoff.load.myRabbitmq.cases.AmqpActions
import ru.tinkoff.load.myRabbitmq.feeders.Feeders.idFeeder
object CommonScenario {
def apply(): ScenarioBuilder = new CommonScenario().scn
}
class CommonScenario {
val scn: ScenarioBuilder = scenario("Common Scenario")
.feed(idFeeder) // вызов Feeder
.exec(AmqpActions.publishMessageToQueue) // выполнение публикации сообщения в очередь
}
Шаг 6. Опишем AMQP-протокол. В файле myRabbitmq.scala опишем протокол:
package ru.tinkoff.load
import io.gatling.core.Predef._
import ru.tinkoff.gatling.amqp.Predef._
import ru.tinkoff.gatling.amqp.protocol.AmqpProtocolBuilder
import ru.tinkoff.gatling.config.SimulationConfig.{getIntParam, getStringParam}
package object myRabbitmq {
val amqpHost: String = getStringParam("amqpHost")
val amqpPort: Int = getIntParam("amqpPort")
val amqpLogin: String = getStringParam("amqpLogin")
val amqpPassword: String = getStringParam("amqpPassword")
val amqpConf: AmqpProtocolBuilder = amqp
.connectionFactory(
rabbitmq
.host(amqpHost)
.port(amqpPort)
.username(amqpLogin)
.password(amqpPassword)
.vhost("/"),
)
.replyTimeout(60000)
.consumerThreadsCount(8)
.usePersistentDeliveryMode
}
Шаг 7. Добавим нагрузочные тесты. В файле Debug.scala опишем тест для дебага, который запустит одну итерацию сценария.
package ru.tinkoff.load.myRabbitmq
import io.gatling.core.Predef._
import ru.tinkoff.gatling.amqp.Predef._
import ru.tinkoff.gatling.config.SimulationConfig._
import ru.tinkoff.load.myRabbitmq.scenarios.CommonScenario
class Debug extends Simulation {
setUp(
CommonScenario() // запускаем наш сценарий
.inject(atOnceUsers(1)), // запускать будет один пользователь - одну итерацию
).protocols(
amqpConf, // работа будет проходить по протоколу, который описан в конфигурации amqpConf
).maxDuration(testDuration) // максимальное время теста равно testDuration,если тест не завершится за меньшее время, он будет остановлен автоматически
}
Шаг 8. Запустим Debug-тест. Для запуска теста возьмем GatlingRunner или способ, указанный в шаге 8. Запуск Debug-теста. После выполнения скрипта можно посмотреть консоль RabbitMQ — количество сообщений в очереди увеличилось на 1.
package ru.tinkoff.load.myRabbitmq
import io.gatling.app.Gatling
import io.gatling.core.config.GatlingPropertiesBuilder
object GatlingRunner {
def main(args: Array[String]): Unit = {
// this is where you specify the class you want to run
val simulationClass = classOf[Debug].getName // указывает имя теста Debug, либо какой-то другой, например, MaxPerformance
val props = new GatlingPropertiesBuilder
props.simulationClass(simulationClass)
Gatling.fromMap(props.build)
}
}
Шаг 9. Настроим публикацию и чтение сообщений. На этом этапе дополнительно добавим чтение сообщений из очереди. В файл AmqpActions запишем новое действие:
val publishAndReply: RequestReplyDslBuilder = amqp("Request Reply exchange test").requestReply
.queueExchange("test_queue_out")
.replyExchange("test_queue_out")
.textMessage("""{"msg": "Hello message - #{messageId}"}""")
.messageId("#{messageId}")
.priority(0)
.contentType("application/json")
.headers("test" -> "performance", "extra-test" -> "34-#{messageId}")
.check(
bodyString.exists,
)
Запишем вызов действия, которое публикует и читает сообщения, в сценарий CommonScenario.
val scn: ScenarioBuilder = scenario("Common Scenario")
.feed(idFeeder) // вызов Feeder
.exec(AmqpActions.publishMessageToQueue) // выполнение публикации сообщения в очередь
.exec(AmqpActions.publishAndReply) // публикация в Exchange и чтение из очереди
Заключение
Это заключительная статья в серии про Gatling. Весь цикл мы рассказывали шаг за шагом, как создать базовые скрипты на Gatling. Надеемся, получилось показать, что на самом деле Gatling — это не так сложно, как кажется на первый взгляд. Если интересно дальше изучать Gatling, рекомендуем Gatling Academy.
Спасибо всем, кто читает наши статьи!