[Перевод] Синхронный Запрос-Ответ с использованием Apache Kafka
Архитектуры, управляемые событиями (Event Driven Architecture), в целом, и Apache Kafka, в частности, привлекли в последнее время большое внимание. Для реализации всех преимуществ архитектуры, управляемой событиями, механизм делегирования событий должен быть по своей сути асинхронным. Тем не менее, могут существовать некоторые особые сценарии/потоки использования, в которых требуется семантика Синхронного Запроса-Ответа. В этом выпуске показано, как реализовать «Запрос-Ответ» с помощью Apache Kafka.
Перевел @middle_java
Дата оригинальной статьи: 26 October 2018
Apache Kafka по своей природе асинхронна. Следовательно, семантика «Запрос-Ответ» для Apache Kafka не является естественной. Тем не менее эта задача не нова. Паттерн Интеграции Корпоративных приложений (Enterprise Integration Pattern) Request-Reply обеспечивает проверенный механизм синхронного обмена сообщениями по асинхронным каналам:
Паттерн Return Address (Адрес Возврата) дополняет паттерн Request-Reply механизмом указания запрашивающей стороной адреса, на который должен быть отправлен ответ:
Недавно в Spring Kafka 2.1.3 была добавлена поддержка из коробки паттерна «Request Reply», а в версии 2.2 были отполированы некоторые из его шероховатостей. Рассмотрим, как работает эта поддержка:
На стороне клиента: ReplyingKafkaTemplate
Хорошо известная абстракция Template (Шаблон) формирует в Spring основу для клиентской части механизма Request-Reply.
@Bean
public ReplyingKafkaTemplate < String, Request, Reply > replyKafkaTemplate(
ProducerFactory < String, Request > pf,
KafkaMessageListenerContainer < String, Reply > lc) {
return new ReplyingKafkaTemplate < > (pf, lc);
}
Здесь все довольно прямолинейно: мы настраиваем ReplyingKafkaTemplate, который отправляет сообщения-запросы со String ключами и получает сообщения-ответы со String ключами. Вместе с тем ReplyingKafkaTemplate должен быть основан на ProducerFactory Запроса, ConsumerFactory Ответа и MessageListenerContainer с соответствующими конфигурациями консюмера и продюсера. Следовательно, необходимая конфигурация довольно развесиста:
@Value("${kafka.topic.car.reply}")
private String replyTopic;
@Bean
public Map < String, Object > consumerConfigs() {
Map < String, Object > props = new HashMap < > ();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
return props;
}
@Bean
public Map < String, Object > producerConfigs() {
Map < String, Object > props = new HashMap < > ();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
return props;
}
@Bean
public ProducerFactory < String, Request > requestProducerFactory() {
return new DefaultKafkaProducerFactory < > (producerConfigs());
}
@Bean
public ConsumerFactory < String, Reply > replyConsumerFactory() {
return new DefaultKafkaConsumerFactory < > (consumerConfigs(), new StringDeserializer(),
new JsonSerializer < Reply > ());
}
@Bean
public KafkaMessageListenerContainer < String, Reply > replyListenerContainer() {
ContainerProperties containerProperties = new ContainerProperties(replyTopic);
return new KafkaMessageListenerContainer < > (replyConsumerFactory(), containerProperties);
}
В этом случае использование replyKafkaTemplate для отправки синхронного запроса и получения ответа выглядит следующим образом:
@Value("${kafka.topic.car.request}")
private String requestTopic;
@Value("${kafka.topic.car.reply}")
private String replyTopic;
@Autowired
private ReplyingKafkaTemplate < String, Request, Reply > requestReplyKafkaTemplate;
...
RequestReply request = RequestReply.request(...);
//создаем producer record
ProducerRecord < String, Request > record = new ProducerRecord < String, Request > (requestTopic, request);
// устанавливаем топик для ответа в заголовке
record.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, requestReplyTopic.getBytes()));
// отправляем запрос в топик Kafka и асинхронно получаем ответ в указанный топик для ответа
RequestReplyFuture < String, Request, Reply > sendAndReceive = requestReplyKafkaTemplate.sendAndReceive(record);
sendAndReceive.addCallback(new ListenableFutureCallback < ConsumerRecord < String, Reply >> () {
@Override
public void onSuccess(ConsumerRecord < String, Reply > result) {
// получаем значение consumer record
Reply reply = result.value();
System.out.println("Reply: " + reply.toString());
}
});
Здесь также много бойлерплейта и низкоуровневого API, да еще этот устаревший API ListenableFuture вместо современной CompletableFuture.
requestReplyKafkaTemplate заботится о генерации и установке заголовка KafkaHeaders.CORRELATION_ID, но мы должны явно задать заголовок KafkaHeaders.REPLY_TOPIC для запроса. Обратите также внимание, что этот же топик для ответа был излишне заинжектен выше в replyListenerContainer. Гадость какая-то. Не совсем то, чего я ожидал от абстракции Spring.
Серверная сторона: @SendTo
На стороне сервера обычный KafkaListener, прослушивающий топик для запроса, дополнительно декорирован аннотацией @SendTo, чтобы предоставить сообщение-ответ. Объект, возвращаемый методом слушателя, автоматически оборачивается (wrapped) в ответное сообщение, добавляется CORRELATION_ID и ответ публикуется в топике, указанном в заголовке REPLY_TOPIC.
@Bean
public Map < String, Object > consumerConfigs() {
Map < String, Object > props = new HashMap < > ();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonSerializer.class);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
return props;
}
@Bean
public Map < String, Object > producerConfigs() {
Map < String, Object > props = new HashMap < > ();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
return props;
}
@Bean
public ConsumerFactory < String, Request > requestConsumerFactory() {
return new DefaultKafkaConsumerFactory < > (consumerConfigs(), new StringDeserializer(),
new JsonSerializer < Request > ());
}
@Bean
public KafkaListenerContainerFactory < ConcurrentMessageListenerContainer < String, Request >> requestListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory < String, Request > factory =
new ConcurrentKafkaListenerContainerFactory < > ();
factory.setConsumerFactory(requestConsumerFactory());
factory.setReplyTemplate(replyTemplate());
return factory;
}
@Bean
public ProducerFactory < String, Reply > replyProducerFactory() {
return new DefaultKafkaProducerFactory < > (producerConfigs());
}
@Bean
public KafkaTemplate < String, Reply > replyTemplate() {
return new KafkaTemplate < > (replyProducerFactory());
}
Здесь также требуется некоторая конфигурация, но конфигурация слушателя проще:
@KafkaListener(topics = "${kafka.topic.car.request}", containerFactory = "requestListenerContainerFactory")
@SendTo()
public Reply receive(Request request) {
Reply reply = ...;
return reply;
}Но как насчет нескольких экземпляров консюмера?
Все вроде работает, пока мы не используем несколько экземпляров консюмера. Если у нас есть несколько экземпляров клиента, мы должны убедиться, что ответ отправляется в корректный экземпляр клиента. Документация Spring Kafka предполагает, что каждый консюмер может использовать уникальный топик или, что с запросом отправляется дополнительное значение заголовка KafkaHeaders.RESPONSE_PARTITION — четырехбайтовое поле, содержащее BIG-ENDIAN-представление целочисленного номера раздела. Использование раздельных топиков для разных клиентов явно не очень гибко, поэтому мы выбираем явную настройку REPLY_PARTITION. Тогда клиент должен знать, на какую партицию он назначен. В документации предлагается использовать явную конфигурацию для выбора конкретной партиции. Давайте добавим ее к нашему примеру:
@Value("${kafka.topic.car.reply.partition}")
private int replyPartition;
...
@Bean
public KafkaMessageListenerContainer < String, RequestReply > replyListenerContainer() {
ContainerProperties containerProperties = new ContainerProperties(replyTopic);
TopicPartitionInitialOffset initialOffset = new TopicPartitionInitialOffset(replyTopic, replyPartition);
return new KafkaMessageListenerContainer < > (replyConsumerFactory(), containerProperties, initialOffset);
}
private static byte[] intToBytesBigEndian(final int data) {
return new byte[] {
(byte)((data >> 24) & 0xff), (byte)((data >> 16) & 0xff),
(byte)((data >> 8) & 0xff), (byte)((data >> 0) & 0xff),
};
}
...
record.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, requestReplyTopic.getBytes()));
record.headers().add(new RecordHeader(KafkaHeaders.REPLY_PARTITION, intToBytesBigEndian(replyPartition)));
RequestReplyFuture < String, RequestReply, RequestReply > sendAndReceive = requestReplyKafkaTemplate.sendAndReceive(record);
...
Не очень красиво, но это работает. Требуемая конфигурация обширна и API выглядит низкоуровнево. Необходимость явной настройки партиций усложняет процесс динамического масштабирования количества клиентов. Очевидно, можно сделать лучше.
Инкапсулирование обработки топика для ответа и партиции
Давайте начнем с инкапсуляции паттерна Return Address, передавая вместе топик для ответа и партицию. Топик для ответа должен быть заинжектен в RequestReplyTemplate и, следовательно, вообще не должен присутствовать в API. Когда речь идет о партициях для ответа, сделаем наоборот: извлечем партицию (партиции), назначенную слушателю топика для ответа, и передадим эту партицию автоматически. Это избавляет клиента от необходимости заботиться об этих заголовках.
При этом, давайте также сделаем API таким, чтобы он напоминал стандартный KafkaTemplate (перегрузим метод sendAndReceive () упрощенными параметрами и добавим соответствующие перегруженные методы, использующие настроенный по умолчанию топик):
public class PartitionAwareReplyingKafkaTemplate < K, V, R > extends ReplyingKafkaTemplate < K, V, R > {
public PartitionAwareReplyingKafkaTemplate(ProducerFactory < K, V > producerFactory,
GenericMessageListenerContainer < K, R > replyContainer) {
super(producerFactory, replyContainer);
}
private TopicPartition getFirstAssignedReplyTopicPartition() {
if (getAssignedReplyTopicPartitions() != null &&
getAssignedReplyTopicPartitions().iterator().hasNext()) {
TopicPartition replyPartition = getAssignedReplyTopicPartitions().iterator().next();
if (this.logger.isDebugEnabled()) {
this.logger.debug("Using partition " + replyPartition.partition());
}
return replyPartition;
} else {
throw new KafkaException("Illegal state: No reply partition is assigned to this instance");
}
}
private static byte[] intToBytesBigEndian(final int data) {
return new byte[] {
(byte)((data >> 24) & 0xff), (byte)((data >> 16) & 0xff),
(byte)((data >> 8) & 0xff), (byte)((data >> 0) & 0xff),
};
}
public RequestReplyFuture < K,
V,
R > sendAndReceiveDefault(@Nullable V data) {
return sendAndReceive(getDefaultTopic(), data);
}
public RequestReplyFuture < K,
V,
R > sendAndReceiveDefault(K key, @Nullable V data) {
return sendAndReceive(getDefaultTopic(), key, data);
}
...
public RequestReplyFuture < K,
V,
R > sendAndReceive(String topic, @Nullable V data) {
ProducerRecord < K, V > record = new ProducerRecord < > (topic, data);
return doSendAndReceive(record);
}
public RequestReplyFuture < K,
V,
R > sendAndReceive(String topic, K key, @Nullable V data) {
ProducerRecord < K, V > record = new ProducerRecord < > (topic, key, data);
return doSendAndReceive(record);
}
...
@Override
public RequestReplyFuture < K,
V,
R > sendAndReceive(ProducerRecord < K, V > record) {
return doSendAndReceive(record);
}
protected RequestReplyFuture < K,
V,
R > doSendAndReceive(ProducerRecord < K, V > record) {
TopicPartition replyPartition = getFirstAssignedReplyTopicPartition();
record.headers()
.add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, replyPartition.topic().getBytes()))
.add(new RecordHeader(KafkaHeaders.REPLY_PARTITION,
intToBytesBigEndian(replyPartition.partition())));
return super.sendAndReceive(record);
}
}
Следующий шаг: Адаптируем ListenableFuture к более современной CompletableFuture.
public class CompletableFutureReplyingKafkaTemplate < K, V, R > extends PartitionAwareReplyingKafkaTemplate < K, V, R > {
public CompletableFutureReplyingKafkaTemplate(ProducerFactory < K, V > producerFactory,
GenericMessageListenerContainer < K, R > replyContainer) {
super(producerFactory, replyContainer);
}
public CompletableFuture < R > requestReplyDefault(V value) {
return adapt(sendAndReceiveDefault(value));
}
public CompletableFuture < R > requestReplyDefault(K key, V value) {
return adapt(sendAndReceiveDefault(key, value));
}
...
public CompletableFuture < R > requestReply(String topic, V value) {
return adapt(sendAndReceive(topic, value));
}
public CompletableFuture < R > requestReply(String topic, K key, V value) {
return adapt(sendAndReceive(topic, key, value));
}
...
private CompletableFuture < R > adapt(RequestReplyFuture < K, V, R > requestReplyFuture) {
CompletableFuture < R > completableResult = new CompletableFuture < R > () {
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
boolean result = requestReplyFuture.cancel(mayInterruptIfRunning);
super.cancel(mayInterruptIfRunning);
return result;
}
};
// добавим коллбек к результату отправки запроса
requestReplyFuture.getSendFuture().addCallback(new ListenableFutureCallback < SendResult < K, V >> () {
@Override
public void onSuccess(SendResult < K, V > sendResult) {
// NOOP
}
@Override
public void onFailure(Throwable t) {
completableResult.completeExceptionally(t);
}
});
// добавим коллбек к ответу
requestReplyFuture.addCallback(new ListenableFutureCallback < ConsumerRecord < K, R >> () {
@Override
public void onSuccess(ConsumerRecord < K, R > result) {
completableResult.complete(result.value());
}
@Override
public void onFailure(Throwable t) {
completableResult.completeExceptionally(t);
}
});
return completableResult;
}
}
Упакуем это в утилитную библиотеку и теперь у нас есть API, который намного больше соответствует основной философии проектирования Spring «Соглашение над Конфигурацией» («Convention over Configuration»). Вот итоговый код клиента:
@Autowired
private CompletableFutureReplyingKafkaTemplate < String, Request, Reply > requestReplyKafkaTemplate;
...
requestReplyKafkaTemplate.requestReply(request).thenAccept(reply - >
System.out.println("Reply: " + reply.toString());
);Подводим итоги
Подводя итог, Spring для Kafka 2.2 обеспечивает полностью функциональную реализацию паттерна Request-Reply в Apache Kafka, но API по-прежнему имеет некоторые шероховатости. В этом выпуске мы увидели, что некоторые дополнительные абстракции и адаптации API могут дать более логичный высокоуровневый API.
Предупреждение 1:
Одним из главных преимуществ архитектуры, управляемой событиями, является разделение (decoupling) продюсеров и консюмеров событий, что позволяет создавать гораздо более гибкие и эволюционирующие системы. Использование синхронной семантики «Запрос-Ответ» является полной противоположностью, когда запрашивающая и отвечающая стороны сильно связаны между собой. Следовательно, ее следует использовать только в случае необходимости.
Предупреждение 2:
Если требуется синхронный Запрос-Ответ, то протокол на основе HTTP намного проще и эффективнее, чем использование асинхронного канала типа Apache Kafka.
Тем не менее, могут быть сценарии, когда синхронный Запрос-Ответ через Kafka имеет смысл. Разумно выбирайте лучший инструмент для работы.
Полностью рабочий пример можно найти на сайте github.com/callistaenterprise/blog-synchronous-kafka.
Комментарии
Federico • 7 месяцев назад
А когда у нас есть гибридные потребности, допустим, в 50% кейсов нам нужен Запрос-Ответ и в 50% нам нужно событийное управление? Как нам это сделать? Конфигурация, необходимая Spring Kafka, выглядит довольно ужасно.
Jehanzeb Qayyum • 6 месяцев назад
Теперь Spring имеет дефолтную поддержку с использованием партиций в одном общем топике для ответа.
Начиная с версии 2.2, шаблон пытается определить топик для ответа или партицию из сконфигурированного контейнера ответа (reply container).
https://docs.spring.io/spring-kafka/reference/html/#replying-template
nir rozenberg • 8 месяцев назад
Привет,
В последнем абзаце вы написали, что могут быть сценарии, когда синхронный Запрос-Ответ через Kafka имеет смысл по сравнению с HTTP.
Можно привести примеры таких сценариев?
Спасибо,
Nir
Janne Keskitalo nir rozenberg • 8 месяцев назад
Мы собираемся разделить систему обработки транзакций большого объема на несколько микросервисов и у меня есть идея использовать обмен сообщениями Kafka «Запрос-Ответ» для достижения похожих способов обработки (processing affinity). В основном Kafka используется для маршрутизации всех вызовов одного клиента в один и тот же процесс обработчика транзакций, который затем последовательно их выполняет по одному. Такой вид обработки гарантирует линеаризируемость (https://stackoverflow.com/a/19515375/7430325), причинно-следственную связность, а также позволяет эффективное кэширование. По сути, усилия по координации были бы перенесены из базы данных в Kafka и мы могли бы запустить базу данных в строгом режиме изоляции Serializable.
Мне еще предстоит углубиться в детали нашей семантики транзакций, чтобы увидеть, где здесь не дотягивает, так что это пока просто идея.
Перевел @middle_java
