Программирование
Java
Apache
Распределённые системы
Микросервисы
18 ноября 2019

Синхронный Запрос-Ответ с использованием Apache Kafka

Автор оригинала: Björn Beskow
Перевод Tutorial
Архитектуры, управляемые событиями (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

+7
3,2k 67
Комментарии 1

Рекомендуем