KafkaProducer: Разница между "обратным вызовом" и возвратом "Будущего"?

Ответ 1

Асинхронный подход

producer.send(record, new Callback(){
    @Override
    onComplete(RecordMetadata rm, Exception ex){...}
})

дает вам лучшую пропускную способность по сравнению с синхронными

RecordMetadata rm = producer.send(record).get();

так как вы не ожидаете подтверждения в первом случае.

Также в асинхронном порядке заказ не гарантируется, тогда как в синхронном - сообщение отправляется только после получения подтверждения.

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

Также обратите внимание, что в асинхронном подходе количество сообщений, которые находятся "во фланге", контролируется параметром max.in.flight.requests.per.connection.

Помимо синхронных и асинхронных подходов вы можете использовать подход Fire and Forget, который почти такой же, как синхронный, но без обработки возвращенных метаданных - просто отправьте сообщение и надейтесь, что он достигнет брокера (зная, что, скорее всего, это произойдет, и продюсер будет повторять попытку в случае восстановления ошибок), но есть вероятность того, что некоторые сообщения будут потеряны:

RecordMetadata rm = producer.send(record);

Подводя итог:

• Fire and Forget - самый быстрый, но некоторые сообщения могут быть потеряны;
• Синхронный - самый медленный, используйте его, если вы не можете позволить себе потерять сообщения;
• Асинхронный - что-то среднее между ними.

Ответ 2

Глядя на документацию, связанную с ней, похоже, что основное различие между Будущим и обратным вызовом заключается в том, кто инициирует "запрос завершен, что теперь?" вопрос.

Скажем, у нас есть клиент C и пекарь B. И C просит B сделать его хорошим печеньем. Теперь есть два возможных способа, которыми пекарь может вернуть вкусный куки файл клиенту.

Future

Бейкер принимает запрос и сообщает клиенту: "Хорошо, когда я закончу, я поместил ваш файл cookie на счетчик. (Это соглашение является Future.)

В этом случае клиент отвечает за проверку счетчика (Future), чтобы узнать, закончил ли пекарь свой файл cookie или нет.

блокировании Клиент остается рядом с прилавком и смотрит на него до тех пор, пока cookie не будет помещен туда (Future.get()), или пекарь вместо этого извиняется (Ошибка: из теста cookie).

неблокирующего Клиент выполняет некоторую другую работу, и время от времени проверяет, ожидает ли его cookie на счетчике (Future.isDone()). Если cookie готов, клиент берет его (Future.get()).

Callback

В этом случае клиент, заказывая свой куки файл, рассказывает пекарю: Когда мой cookie готов, пожалуйста, отдайте его моей собаке-роботу-собаке здесь, он будет знать, что с ним делать (этот робот - Обратный звонок).

Теперь пекарь, когда cookie готов, дает печенье собаке и говорит ему, чтобы он вернулся к нему. Пекарь может продолжить выпечку следующего файла cookie для другого клиента.

Собака возвращается к клиенту и начинает вилять искусственный хвост, чтобы клиент знал, что его cookie готов.

Обратите внимание, что клиент не имел ни малейшего представления о том, когда ему будет дано куки файл, и он не стал активно опросить пекаря, чтобы убедиться, что он готов.

Это основное отличие между сценарием 2. Кто несет ответственность за инициирование "ваш cookie готов, что вы хотите с ним делать?" вопрос. В будущем клиент несет ответственность за проверку, когда он готов, либо активно ожидая, либо опросив время от времени. В случае обратного вызова пекарь вернется к предоставленной функции.

Я надеюсь, что этот ответ даст вам лучшее представление о том, что такое "Будущее и Калбек". Как только вы получите общую идею, вы можете попытаться выяснить, в каком потоке обрабатывается каждая конкретная вещь. Когда поток заблокирован или в каком порядке все завершается. Написание некоторых простых программ, которые печатают такие заявления, как: "main client thread: cookie received" может быть интересным способом поэкспериментировать с этим.

Ответ 3

Основное различие заключается в том, хотите ли вы заблокировать вызывающий поток, ожидающий подтверждения.

Следующее использование метода Future.get() блокирует текущий поток до тех пор, пока передача не будет выполнена до выполнения каких-либо действий.

producer.send(record).get()
// Do some action

При использовании обратного вызова для выполнения некоторых действий код будет выполняться в потоке ввода-вывода, чтобы он не блокировал вызывающий поток.

 producer.send(record,
               new Callback() {
                   // Do some action
                   }
               });

Хотя документы говорит, что он "вообще" выполнен в роли производителя:

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

Ответ 4

Мои наблюдения, основанные на Документация производителя Kafka:

• Future предоставляет вам доступ к синхронной обработке
• Future может не гарантировать подтверждения. Я понимаю, что Callback будет выполняться после подтверждения
• Callback предоставляет вам доступ к асинхронной обработке полностью неблокирующейся.
  • Существуют также гарантии при заказе выполнения для обратного вызова в том же разделе

Гарантируются обратные вызовы для записей, отправляемых на один раздел выполнить в порядке.

Мое другое мнение о том, что возвращаемый объект Future и шаблон Callback 'представляют два разных стиля программирования, и я думаю, что это фундаментальное различие:

• Future представляет стиль стиля Java Concurrency.
• Callback представляет собой Java Lambda Programming Style (потому что Callback действительно удовлетворяет требованиям для функционального интерфейса)

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

Ответ 5

send() - это метод начала публикации сообщения в кластере Kafka. Метод send() - это асинхронный вызов, который говорит, что метод send накапливает сообщение в буфере и немедленно возвращает обратно. Это можно использовать с linger.ms для пакетной публикации сообщений для повышения производительности. Мы можем обрабатывать исключения и управлять с помощью метода отправки вызова с синхронным использованием метода get в Future или асинхронного с обратным вызовом.

Каждый метод имеет свои плюсы и минусы и может быть решен на основе вариантов использования.

Асинхронная отправка (Fire & Forget): Мы вызываем метод отправки, как показано ниже, чтобы вызвать публикацию сообщения без ожидания какого-либо успеха или ответа об ошибке.

producer.send(new ProducerRecord<String, String>("topic-name", "key", "value"));

Этот сценарий не будет ждать завершения первого сообщения, начните отправку другого сообщения для публикации. В случае возникновения исключительной ситуации производитель повторяет попытку на основании параметра конфигурации retry, но если сообщение все еще не выполнено после повторной попытки, Kafka Producer никогда не узнает об этом. В этом случае мы можем разместить несколько сообщений, но если все в порядке с небольшой потерей сообщений, это обеспечит высокую пропускную способность и высокую задержку.

Синхронная отправка Простой способ отправить сообщение синхронно - использовать метод get()

RecordMetadata recMetadata = producer.send(new ProducerRecord<String, String>("topic-name", "key", "value")).get();

Producer.send возвращает Future of RecordMetadata, и когда мы вызываем метод .get(), он получает ответ от Kafka. Мы можем поймать Error в случае ошибки или вернуть RecordMetadata в случае успеха. RecordMetadata содержит смещение, раздел, метку времени для регистрации информации. Это медленно, но дает высокую надежность и гарантирует доставку сообщения.

Асинхронная отправка с обратным вызовом Мы также можем вызвать метод send() с функцией обратного вызова, которая возвращает ответ после завершения сообщения. Это хорошо, если вы хотите отправить сообщение асинхронным способом, то есть не ждать завершения работы, а одновременно обрабатывать ошибку или обновлять статус доставки сообщения.

producer.send(record, new Callback(){
    @Override
    onComplete(RecordMetadata recodMetadata, Exception ex){...}
})

Примечание: Пожалуйста, не путайте с ack & повторяет попытку с асинхронным вызовом отправки. Подтверждение и повторные попытки будут применяться к каждому отправляющему вызову, будь то его синхронный или асинхронный вызов, единственное, как вы обрабатываете возвратные сообщения и сценарий сбоя. Например, если вы отправляете асинхронную отправку, все еще подтверждаете, и правило повторных попыток применяется, но оно будет в независимом потоке, не блокируя другой поток для отправки параллельных записей. Единственная проблема, которую мы не будем знать в случае сбоя и время, когда сообщение успешно завершено.