Существует множество способов обработки сообщений из Pub-Sub систем: использование отдельного сервиса, выделение изолированного процесса, оркестрация пулом процессов/потоков, сложные IPC, Poll-over-Http и многие другие. Сегодня я хочу рассказать о том, как использовать Pub-Sub по HTTP и про свой сервис, написанный специально для этого.

Использование готового HTTP -бэкенда сервисов в некоторых случаях является идеальным решением для обработки очереди сообщений:

1. Балансировка из коробки. Обычно, бэкенд и так стоит за балансировщиком и имеет готовую к нагрузкам инфраструктуру, что сильно упрощает работу с сообщениями.
2. Использование обычного REST-контроллера (любой HTTP-ресурс). Потребление сообщений по HTTP сводит к минимуму затраты на реализацию консюмеров под разные языки, если бэкенд разношерстный.
3. Упрощение использования Веб-хуков других сервисов. Сейчас почти каждый сервис (Jira, Gitlab, Mattermost, Slack…) так или иначе поддерживает Веб-хуки для взаимодействия с внешним миром. Можно облегчить жизнь, если научить очередь выполнять функции HTTP-диспатчера.

Этот подход имеет и минусы:

1. Можно забыть о легковесности решения. HTTP тяжёлый протокол, а использование фреймворков на стороне консюмера мгновенно приведёт к увеличению задержки (latency) и нагрузки.
2. Лишаемся сильных сторон Poll-подхода, получая слабые стороны Push.
3. Обработка сообщений теми же инстансами сервиса, которые обрабатывают клиентов, может сказаться на отзывчивости. Это несущественно, так как лечится балансировкой и изоляцией.

Я реализовал идею в виде сервиса Queue-Over-Http, о котором и пойдёт речь далее. Проект написан на Kotlin с использованием Spring Boot 2.1. В качестве брокера сейчас доступна только Apache Kafka.

Далее в статье подразумевается, что читатель знаком с Kafka и знает про коммиты (commit) и оффсеты (offset) сообщений, принципы групп (group) и консюмеров (consumer), а также понимает, чем партиция (partition) отличается от топика (topic). Если есть пробелы, советую ознакомиться с этим разделом документации по Kafka перед продолжением чтения.

Содержание

• Обзор
• Коммиты
• Обработка ошибок
• Сообщения
• Производительность
• Демонстрация
• Заключение

Обзор

Queue-Over-Http представляет из себя сервис, который выступает посредником между брокером сообщений и конечным HTTP-консюмером (сервис позволяет легко реализовать поддержку отправки сообщений консюмерам любым другим способом, например, различными *RPC). На данный момент доступны только операции подписки, отписки и просмотра списка консюмеров Отправка сообщений брокеру (produce) по HTTP пока не реализована в силу невозможности гарантировать порядок сообщений без специальной поддержки со стороны продюсера.

Ключевой фигурой сервиса является консюмер, который может подписаться как на конкретные партиции, так и просто на топики (паттерн топика поддерживается). В первом случае выключается автобаланс партиций. После подписки, указанный HTTP-ресурс начинает получать сообщения из назначенных партиций Kafka. Архитектурно каждый подписчик ассоциируется с нативным Java-клиентом Kafka.


Каждая партиция представлена отдельной локальной очередью, куда адаптер пишет сообщения из брокера. Воркер забирает из неё сообщения и отдаёт их на исполнение, то есть, на отправку по HTTP.

Сервис поддерживает пакетную обработку сообщений для увеличения пропускной способности. При подписке можно указать concurrencyFactor каждого топика (распространяется на каждую назначенную партицию независимо). Например, concurrencyFactor=1000 означает, что одновременно могут быть отправлены потребителю 1000 сообщений в виде HTTP-запросов. Как только все сообщения из пачки были однозначно отработаны консюмером, сервис принимает решение об очередном коммите оффсета последнего по порядку сообщения в Kafka. Отсюда второе значение concurrencyFactor — максимальное число повторно обработанных сообщений потребителем в случае падения Kafka или Queue-Over-Http.

Для уменьшения задержек очередь имеет loadFactor = concurrencyFactor * 2, что позволяет считывать из брокера в два раза больше сообщений, чем может быть отправлено. Так как автокоммит на нативном клиенте отключен, такая схема не нарушает гарантий At-Least-Once.
Высокое значение concurrencyFactor увеличивает пропускную способность очереди (throughput) за счёт уменьшения количества коммитов, которые занимают до 10 мс в худшем случае. При этом, повышается нагрузка на потребителя.

Очерёдность отправки сообщений в рамках пачки не гарантирована, но её можно достигнуть, если установить concurrencyFactor=1.

Коммиты

Коммиты — важная часть работы сервиса. Когда очередная пачка данных готова, оффсет последнего сообщения из пачки тут же коммитится в Kafka, и только после успешного коммита становится доступна для обработки следующая пачка. Часто этого недостаточно и требуется автокоммит. Для этого существует параметр autoCommitPeriodMs, который имеет мало общего с классическим периодом автокоммита у нативных клиентов, которые коммитят последнее прочитанное из партиции сообщение. Представим, что concurrencyFactor=10. Сервис отослал все 10 сообщений и ждёт готовности каждого из них. Первым завершается обработка сообщения 3, потом сообщения 1, а затем, сообщения 10. В этот момент наступает время автокоммита. Важно не нарушить At-Least-Once семантику. Поэтому, можно коммитить только первое сообщение, то есть, оффсет 2, так как только оно на этот момент успешно обработано. Далее, до следующего автокоммита обрабатываются сообщения 2, 5, 6, 4, и 8. Теперь необходимо коммитить только оффсет 7, и так далее. Автокоммит почти не влияет на пропускную способность.

Обработка ошибок

В штатном режиме работы сервис отправляет сообщение консюмеру один раз. Если по каким-то причинам оно вызвало 4xx или 5xx ошибку, то сервис будет повторно отправлять сообщение, ожидая успешной обработки. Время между попытками может быть сконфигурировано отдельным параметром.

Так же, возможно задать количество попыток, по истечению которых сообщение будет помечено как обработанное, что прекратит повторные отправки не зависимо от статуса ответа. Не советую использовать это для чувствительных данных, ситуации отказа консюмеров всегда нужно корректировать вручную. Залипания сообщения можно отслеживать по логам сервиса и мониторингу статусов ответа консюмера.

Сообщения

Каждое сообщение, извлечённое из брокера, отправляется консюмеру по HTTP на указанный при подписке ресурс. По умолчанию сообщение отправляется POST-запросом в теле. Это поведение можно изменить, указав любой другой метод. Если метод не поддерживает отправку данных в теле, можно указать название строкового параметра, в котором будет отправлено сообщение. Помимо этого, при подписке можно указать дополнительные заголовки, которые будут добавлены к каждому сообщению, что удобно для базовой авторизации с помощью токенов. К каждому сообщению добавляются заголовки с указанием идентификатора консюмера, топика и партиции, откуда сообщение было прочитано, номер сообщения, partition key, если применим, а также название самого брокера.

Производительность

Для оценки производительности я использовал ПК (Windows 10, OpenJDK-11 (G1 без тюнинга), i7-6700K, 16GB), на котором запущен сервис и ноутбук (Windows 10, i5-8250U, 8GB), на котором крутился продюсер сообщений, HTTP-ресурс консюмера и Kafka с дефолтными настройками. ПК подключен к роутеру по проводному соединению 1Gb/s, ноутбук по 802.11ac. Продюсер каждые 100 мс в течении 1000 секунд записывает сообщения, длиной в 110 байт, в назначенные топики, на которые подписаны консюмеры (concurrencyFactor=500, автокомит выключен) из разных групп. Стенд далёк от идеального, но некоторую картину получить можно.

Ключевым измеряемым параметром является влияние сервиса на latency.

Пусть:
— t_q – временная метка получения сервисом сообщения от нативного клиента
— d_t0 – время между t_q и временем отправки сообщения из локальной очереди в пул экзекьютеров
— d_t – время между t_q и временем отправки HTTP-запроса. Именно d_t является влиянием сервиса на latency сообщения.

В ходе измерений были получены следующие результаты (C – консюмеры, T – топики, M — сообщения):



В стандартном режиме работы сервис сам по себе почти не влияет latency, а потребление памяти минимально. Максимальные значения d_t (около 60мс) не указаны специально, так как зависят от работы GC, а не от самого сервиса. Сгладить разброс максимальных значений может помочь специальный тюнинг GC или замена G1 на Shenandoah.

Всё кардинально меняется, когда консюмер не справляется с потоком сообщений из очереди и сервис включает режим тротлинга. В этом режиме увеличивается потребление памяти, так как время ответа на запросы сильно вырастает, что мешает своевременной очистке ресурсов. Влияние на latency здесь остаётся на уровне с предыдущими результатами, а высокие значения dt вызваны предзагрузкой сообщений в локальную очередь.

К сожалению, протестировать на бОльшей нагрузке не имеется возможности, так как ноутбук загибается уже на 1300 RPS. Если кто-то может помочь с организацией замеров на больших нагрузках, с радостью предоставлю сборку для тестов.

Демонстрация

Теперь перейдём к демонстрации. Для этого нам понадобится:

• Kafka брокер, готовый к работе. Я возьму поднятый на 192.168.99.100:9092 инстанс от Bitnami.
• HTTP-ресурс, который будет принимать сообщения. Для наглядности я взял Web-hooks у Slack.

Прежде всего, необходимо поднять сам сервис Queue-Over-Http. Для этого создим в пустой директории application.yml следующего содержания:

spring:
  profiles: default

logging:
  level:
    com:
      viirrtus:
        queueOverHttp: DEBUG

app:
  persistence:
    file:
      storageDirectory: "persist"
  brokers:
    - name: "Kafka"
      origin: "kafka"
      config:
        bootstrap.servers: "192.168.99.100:9092"

Здесь мы указываем сервису параметры подключения конкретного брокера, а также, где хранить подписчиков, чтобы между запусками они не терялись. В `app.brokers[].config` можно указывать любые, поддерживаемые нативным клиентом Kafka параметры подключения, полный список можно посмотреть здесь.

Так как файл конфигурации обрабатывается Spring’ом, вы можете писать туда много интересного. В том числе, настраивать логирование.

Теперь запускаем сам сервис. Используем самый простой способ – docker-compose.yml:

version: "2"

services:
  app:
    image: viirrtus/queue-over-http:0.1.3
    restart: unless-stopped
    command: --debug
    ports:
      - "8080:8080"
    volumes:
      - ./application.yml:/application.yml
      - ./persist:/persist

Если этот вариант не устраивает, вы можете собрать сервис из исходников. Инструкция по сборке в Readme проекта, ссылка на который дана в конце статьи.

Следующим шагом регистрируем первого подписчика. Для этого необходимо выполнить HTTP-запрос к сервису с описанием консюмера (Consumer):

POST localhost:8080/broker/subscription
Content-Type: application/json

{
  "id": "my-first-consumer",
  "group": {
    "id": "consumers"
  },
  "broker": "Kafka",
  "topics": [
    {
      "name": "slack.test",
      "config": {
        "concurrencyFactor": 10,
        "autoCommitPeriodMs": 100
      }
    }
  ],
  "subscriptionMethod": {
    "type": "http",
    "delayOnErrorMs": 1000,
    "retryBeforeCommit": 10,
    "uri": "<slack-wh-uri>",
    "additionalHeaders": {
      "Content-Type": "application/json"
    }
  }
}

Если всё прошло успешно, в ответе будет почти тот же самый отправленный контент.

Пройдёмся по каждому параметру:

• Consumer.id — идентификатор нашего подписчика
• Consumer.group.id — идентификатор группы
• Consumer.broker — указываем на какой из брокеров сервиса нужно подписаться
• Consumer.topics[0].name — название топика, из которого хотим получать сообщения
• Consumer.topics[0].config. concurrencyFactor — максимальное количество одновременно отправленных сообщений
• Consumer.topics[0].config. autoCommitPeriodMs — период принудительного коммита готовых сообщений
• Consumer.subscriptionMethod.type — тип подписки. В данный момент доступна только HTTP.
• Consumer.subscriptionMethod.delayOnErrorMs — время до повторной отправки сообщения, которое закончилось ошибкой
• Consumer.subscriptionMethod.retryBeforeCommit — количество попыток повторной отправки ошибочного сообщения. Если 0 – сообщение будет крутиться до успешной обработки. В нашем случае гарантия полной доставки не так важна, как постоянство потока.
• Consumer.subscriptionMethod.uri — ресурс, на который будут отправляться сообщения
• Consumer.subscriptionMethod.additionalHeader — дополнительные заголовки, которые будут отправлены с каждым сообщением. Пометим, что в теле каждого сообщения будет JSON, чтобы Slack мог правильно интерпретировать запрос.

В данном запросе указание HTTP-метода опущено, так как умолчание, POST, Slack вполне устраивает.

С этого момента сервис следит за назначенными партициями топика slack.test на предмет новых сообщений.

Для записи сообщений в топик я воспользуюсь встроенным в Kafka утилитами, которые расположены в /opt/bitnami/kafka/bin запущенного образа Kafka (расположение утилит в других инстансах Kafka может отличаться):

kafka-console-producer.sh --broker-list localhost:9092 --topic slack.test
> {“text”: “Hello!”}

В этот же момент Slack оповестит о новом сообщении:



Чтобы отписать потребителя достаточно сделать POST запрос на `broker/unsubscribe` с тем же контентом, что был при подписке.

Заключение

На данный момент реализован лишь базовый функционал. Далее планируется улучшить batching, попытаться реализовать Exactly-once семантику, добавить возможность отправки сообщений брокеру по HTTP и, самое главное, добавить поддержку других популярных Pub-Sub.

Сервис Queue-Over-Http сейчас находится в стадии активного развития. Версия 0.1.3 является достаточно стабильной для тестирования на dev и stage стендах. Работоспособность была проверена на Windows 10, Debian 9 и Ubuntu 18.04. Использовать в prod можно на свой страх и риск. Если вы хотите помочь с разработкой или дать любой фидбэк по сервису – добро пожаловать на Github проекта.