что происходит с сообщениями, когда соединение между сервером kafka и производителем нарушено?

Question

Я новичок в Кафке, используя Spring Boot, и я работаю в ProJet. Я хочу интегрировать в него Кафку, используя Spring, поэтому проблема в том, что я хочу отправить сообщение от производителя потребителю, даже если Сервер Kafka не (автономный режим)

Может ли кто-нибудь дать мне пример использования Кафки в автономном режиме, я не могу найти туто по этой теме, я хочу остановить свой сервер Кафки (например)и в то же время производитель хочет отправить данные в тему, тогда потребитель может получить эти сообщения?Какое лучшее решение? Это правда?

* отправка данных в файл, и когда сервер возвращается к работе (я проверяю подключение, например), я экспортирую данные из файла в тему

* отправка данных в базу данных, и когда сервер возвращается к работе (проверка соединения), то же самое я отправляю сообщение (данные) из базы данных в мою тему

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

-> если есть другое решение с простым примером, может кто-нибудьПомогите мне?

Это пример брокера Redis , который мы проверяем соединение между Redis Broker и производителем, если соединение не удается, я буду хранить данные в Очередь , котораяможет хранить много сообщений, и когда соединение возвращается к работе между Redis и производителем, производитель теперь получает эти сообщения из очереди и отправляет их в Redis Brocker.

Но проблемаВ этом брокере потеряно несколько сообщений, поэтому мы решили интегрировать брокер kafka в мой проект вместо Redis brocker!

Может ли кто-нибудь дать мне пример в java, как хранить много сообщений перед их отправкойпроизводитель к кластеру kafka? или каково лучшее решение этой проблемы, потому что мы не хотим использовать одно и то же решение очереди?

этот пример в python - как сохранить сообщение в очереди, если подключение не удалось к серверу:

    try:
    urllib.request.urlopen('http://serverAdress', timeout=0.1)
    r.publish(topicProduction,json_background_of_message1)
    print(json_background_of_message1)
    arretControle=Tru
    except Exception as e:
    qArret.put(json_background_of_message1)
    print("arret")
    arretControle=True

// json_background_of_message1 - это очередь, в которой мы можем хранить много сообщений в этой очереди перед отправкой этих сообщений, если соединение не удалось

Answer 1

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

То, что вы предлагаете, - это запасной вариант или метод резервного копирования на случай, если Kafka недоступен.Тем не менее, у вас все еще есть та же проблема.Если вы дампите сообщения в файл, как долго, пока вы не исчерпаете место на диске?Если вы храните сообщения в базе данных, как долго, пока база данных не исчерпает пространство?Если вы храните сообщения в очереди в памяти, сколько времени пройдет, пока у вас не закончится память и не произойдет сбой приложения?И теперь вам также необходимо создать механизм восстановления после сбоя kafka, который добавляет сложность и накладные расходы.

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

Теперь на стороне приложения ваше поведение вслучай, когда Кафка недоступна, должен зависеть от того, насколько важны сообщения.Если вы можете терпеть потерю сообщений, просто отбросьте их, если производитель возвращает исключение, и зарегистрируйте его / отправьте предупреждение.Однако, если они являются чрезвычайно важными записями, вы не должны подтверждать / фиксировать сообщения в вашей исходной системе (откуда бы ни исходили записи), пока не получите полное подтверждение того, что они сохранены в Kafka.Я бы порекомендовал установить для производителя подтверждение -1 или all, несколько повторных попыток в случае сбоя и настройку правильного метода обратного вызова для метода producer.send().Более подробное объяснение см. Здесь: https://kafka.apache.org/21/javadoc/index.html?org/apache/kafka/clients/producer/Callback.html

Для получения более подробной информации, как сказали другие, прочитайте официальные документы: https://kafka.apache.org/documentation/