Ручное сообщение подтверждает в Kafka…
Apache Kafka — это распределенная система обмена сообщениями с публикацией и подпиской, которая собирает данные из разных исходных систем и делает их доступными для целевых систем в режиме реального времени. В наши дни Kafka является популярным потоковым брокером. давайте обсудим, как разработать решение с помощью Kafka и как практически работает Kafka.
Здесь, в варианте использования 01, одиночный потребитель принимает сообщения из всех этих трех разделов. Подумайте, нам нужно масштабировать это приложение, поэтому мы добавляем еще одного потребителя для этой конкретной группы, давайте разберемся с этим,
Теперь, согласно сценарию варианта использования 02, потребитель 2 является точкой для раздела 2, а потребитель 1 — точкой для разделения 0 и 1. Этот порядок полностью случайный. Этот порядок можно изменить в зависимости от того, как координируются слушатели и темы. Неважно, как этот порядок работает, как вариант использования 02. Когда мы добавим еще одного потребителя для этой группы потребителей и нам нужно снова масштабировать приложение, что произойдет? давайте обсудим это.
Видите, теперь каждый потребитель читает из каждого раздела. Подумайте, когда мы добавим еще одного потребителя для этой группы потребителей, что произойдет? Что вы думаете? это ускорит процесс? давайте обсудим это,
Kafka не будет отправлять сообщения потребителю 04, что означает, что потребитель 4 будет бездействовать. Когда вы добавляете больше, эти потребители также простаивают. Почему это? У Кафки есть правило. у вас может быть максимальное количество потребителей в группе потребителей по количеству разделов, которые у нас есть. согласно варианту использования 04, потребитель 04 важен, когда потребитель 3 или 2 каким-то образом вышел из строя, Kafka немедленно передает остальные вещи следующему потребителю 04 для этого раздела. Давайте обсудим еще один вариант использования.
Когда мы добавляем еще одну группу потребителей для той же темы, 1-я группа потребителей имеет 3 потребителя, указывающих на каждый раздел. но в группе потребителей 2 (финансово-потребительская группа) у вас есть только 2 потребителя. Итак, потребитель 2 прослушивает два раздела, а потребитель 1 прослушивает раздел 1. У нас проблема, верно? Как этот потребитель узнает, откуда я беру сообщения и так далее, давайте это обсудим. Кафка поддерживает удержание, утверждали они. мы можем назвать это Смещение потребителя, основываясь на том, что Кафка точно знает этого потребителя прямо отсюда, и ему нужно перейти сюда и все такое. , количество групп потребителей должно быть меньше или равно количеству разделов в теме.
В этом коде мы собираемся создать тему с 3 разделами. имя темы — тема сотрудника, и она была создана в соответствии с этим кодом.
В этом коде мы должны импортировать Kafka из Kafka.js, и здесь мы создали часть подключения и брокера, работающего на локальном порту 9092. (01)
Затем нам нужно получить наш продюсер. (02)
Метод публикации заставляет производителя подключиться, а затем в цикле for мы использовали эту тему сотрудника, а затем передавали это простое сообщение. (03)
Затем мы создали файл Consumer.js. Сначала мы настроили соединение. (01)
Затем мы создали нашего потребителя, выслушав ту же тему «тема сотрудника». (02)
Когда мы получаем сообщение, мы устанавливаем раздел и смещение сообщения, а также значение сообщения.(03)
Затем мы запускаем этот файл Consumer.js с помощью команды узла (01), и здесь вы можете увидеть результат (02), {“employee-topic”: [0,1,2]}, что означает, что все 3 созданных нами раздела прослушиваются этим потребителем. это сценарий UseCase 01.
Затем мы создаем другого потребителя, в этом коде ясно видно, что он прослушивает только разделы [0,2]. то теперь что будет с первым потребителем посмотрим
Теперь вы можете видеть, что этот потребитель слушает только раздел 01, это сценарий UseCase 02. давайте создадим еще одного потребителя,
Посмотрите, теперь этот потребитель слушает Partition 0. что будет с потребителем 1 и 2, посмотрим
Во втором потребителе вы можете увидеть его перебалансировку на раздел 1, а первый потребитель перебалансирует его на раздел 02. Теперь вы можете понять, что всякий раз, когда вы добавляете потребителя в группу потребителей, Kafka автоматически балансирует на основе количества разделов, которые у нас есть. Итак, если вы добавите еще одного потребителя в группу потребителей, что произойдет? ничего не произойдет, и этот потребитель будет простаивать.
Теперь запускаем производителя.js с помощью команды node, посмотрим, что получится,
Первый потребитель получил 4 сообщения. Второй потребитель получил 2 сообщения, а последний потребитель получил 4 сообщения. вы можете видеть, что 10 сообщений сбалансированы. Балансировка нагрузки между потребителями.
Давайте обсудим некоторые расширенные варианты использования Kafka…
Пакетный процесс
Вместо этого eachMessage мы можем использовать метод eachBatch. Это пакетный обработчик. каждое сообщение обрабатывает сообщения одно за другим, а каждый пакет может обрабатывать набор сообщений как пакет. мы можем использовать этот метод eachBatch вместо eachMessage внутри метода consumer.run в соответствии с этим кодом.
В eachBatch у нас больше возможностей, чем в eachMessage. он имеет resolveOffset, heartbeat, uncommittedOffsets.
Автофиксация
Например, если вы получите 500 сообщений одновременно и если каким-то образом 400 сообщений будут разбиты, что произойдет? затем он отправит все снова, потому что вы не отправили коммит. чтобы избежать этой проблемы, мы можем использовать свойство с именем autoCommitThreshold. это означает, что если мы установим пороговое значение в 10 сообщений, оно отправит зафиксированное смещение обратно на сервер. это означает, что каждые 10 сообщений отправляются на сервер, иначе у нас есть другое свойство с именем autocommitInterval. в зависимости от продолжительности, в течение которой он отправит зафиксированное смещение на сервер.
Кроме того, мы можем установить для этого свойства autoCommit значение true или false. когда для autoCommit установлено значение false, Kafka не будет автоматически обрабатывать часть фиксации, вам нужно вручную обрабатывать процесс фиксации. это означает, что когда вы получили сообщение, вы выполняете свою задачу, и если она успешна, вы можете сказать «Я закончил с этим» или что-то еще и зафиксировать ее. если вы не зафиксировали, зафиксированное смещение останется.
Время ожидания сеанса
Это свойство важно, так как оно управляет процессом перебалансировки. в качестве примера мы предполагаем, что время ожидания сеанса составляет 30 секунд. если вы не можете отправить сердцебиение в течение 30 секунд. этот потребитель отсоединится от Кафки. если ваш процесс занимает больше времени, чем тайм-аут сеанса, ваш сигнал пульса не достигает сервера, и в результате вы будете освобождены от кластера. чтобы избежать этого, у нас есть 2 решения: нам нужно увеличить sessionTimeout или сохранить sessionInterval в надежном количестве и каждое сообщение принимает пульс в качестве входных данных(01) и все, что можно назвать пульсом. (02)
Использование этого тактового импульса сообщит Kafka, что потребитель активен, и таким образом вы можете максимально уменьшить sessionTimeout, а процесс может продолжаться. кластер и избегайте отсоединения.
Спасибо!
Использованная литература :
