Опрашивая Kafka, я подписался на несколько тем, используя функцию subscribe(). Теперь я хочу установить смещение, с которого я хочу читать из каждой темы, без повторной подписки после каждых seek() и poll() из темы. Достигнет ли результата итеративный вызов seek() по каждому из названий тем, перед опросом данных? Как именно смещения хранятся в Кафке?
У меня есть один раздел для каждой темы и только один потребитель для чтения по всем темам.
Как Kafka хранит смещения для каждой темы?
Kafka переместил офсетное хранилище от zookeeper к брокерам kafka. Причина ниже:
Zookeeper не является хорошим способом обслуживания высокой нагрузки записи, такой как обновления смещения, потому что zookeeper направляет каждую запись через каждый узел и, следовательно, не имеет возможности разбивать или иным образом масштабировать записи. Мы всегда знали об этом, но выбрали эту реализацию как своего рода «брак по расчету», поскольку мы уже зависели от zk.
Kafka сохраняет смещение, фиксируемое в теме, когда потребитель фиксирует смещение, kafka публикует сообщение о смещении фиксации в тему «журнала фиксации» и сохраняет структуру в памяти, которая сопоставляет группу / тему / раздел с последним смещением для быстрого поиска. . Дополнительную информацию о дизайне можно найти на этой странице об управлении смещением.
Теперь я хочу установить смещение, с которого я хочу читать из каждой темы, без повторной подписки после каждого seek () и poll () из темы.
В инструментах администрирования kafka появилась новая функция для сброса смещения.
kafka-consumer-group.sh --bootstrap-server 127.0.0.1:9092 --group
your-consumer-group **--reset-offsets** --to-offset 1 --all-topics --execute
У вас есть другие варианты. можно использовать.
