Comments 47
Ну... центральное место наверное скорее - хранилище :) а то я уж по заголовку подумал что в самой кафке хранить научились, типа как оперативная память была на звуковых волнах.
А если бы не кафка была, все поломалось бы? :)
я не совсем вас понял. Kafka вполне себе выступает как хранилище, в топике хранится реплика данных из postgres.
А если бы не кафка была, все поломалось бы? :) - тоже не понял что имеется ввиду, что значит была или не была, у нас уже используется кафка, почему и как - это отдельная наверное дискуссия должна быть, кафка как сказано в начале является центральной высокодоступной системой и если она не работает, то да, все не работает - такое допущение, при этом если не работает postgres - часть системы должна функционировать
Я тоже разделяю смятение, которое выразил пользователь выше. Тоже по заголовку подумал, что вы в кафке данные научились хранить. Но по сути у вас есть сервис, который с помощью Кафки доставляет данные, но никак не хранит. А хранит он в оперативной памяти. И тогда возникает вопрос, что делать если сервис упал- поднялся и в оперативной памяти данных нет, а в кафке топики уже прочитаны?
а я не понимаю вашего смятения:)
данные именно что хранятся в 1) postgres 2) в compacted топике кафка 3) в оперативной памяти сервиса. Не знаю возможно и можно придумать название удачней, но я вас точно не обманул, данные вполне себе хранятся в топике кафка и кафка вполне себе выступает хранилищем из которого данные вытягиваются в приложение.
если сервис упал-поднялся-отжался - то все данные при инициализации сервиса будут загружены с самого начала. Посмотрите внимательней код KafkaGlobalStore, также я отдельно расписал про проблемы, что нужно тестировать и оценивать размер этих справочников, чтобы для вас не было дорого как хранение данных в оперативной памяти так и их загрузка каждый раз туда при старте сервиса. Посмотрите разделы "Потребляемые ресурсы и время загрузки" и "Что делать, если появятся большие справочники". При каждом старте приложения топик будет вычитываться сначала, также я ответил на комментарий ниже про горизонтальное масштабирования, смысл тот же - для каждого инстанса при каждом старте топик будет вычитываться заново до старта основной логики в идеале ( тут уж spring в помощь)
Но ведь у Кафка топиков есть ttl, не все данные можно будет загрузить сначала
у кафка топик есть разные retention policy и мы используем compacted для таких топиков. Тоже упомянул про это в статье со ссылкой:
Таким образом любой сервис, который слушает Kafka topic, может собрать у себя актуальную реплику справочника. Но нужно не забывать, чтобы key(row i) не изменялся. В таком случае мы можем применить к kafka topic retention policy compacted. Т.к. нас интересует только последний record по ключу пусть kafka удаляет старые данные для одинаковых ключей в соответствии с retention policy.
Но зачем другим сервисам собирать реплику, если это делает ваш сервис и представляет об этом данные?
Но и опять таки, вопрос то был про ttl. То есть часть данных просто будет недоступна и в итоге невозможно будет восстановить реплику полностью. Плюс размер топика.
В общем, я понимаю, что это все возможно сделать, пару лет назад я сам делал систему частичной репликации постгри через кафку, но одновременно с этим я помню насколько это мучительно , потому что есть инструменты более подходящие
к сожалению не понял в такой формулировке вашего вопроса, но попробую додумать и ответить на примере, который приведен
Есть справочник terminal_acquirer. Служба поддержки вводит 100 новых терминалов, первые 50 должны идти в Сбер, вторые в Альфа. Далее у нас есть сервис роутер, в которые приходит транзакция в которой есть terminal_id. С помощью реплики справочник в сервисе мы определяем куда дальше отправлять сообщение или делать запрос.
По ttl тоже непонятно, я вам вроде ответил, что для cleanup.policy=compact нет ttl - перейдите по ссылке.
Про репликацию постгри с вами подискутировать не могу, у меня не было такой задачи и не знаю какая именно задача была у вас и что вы имеете ввиду под более подходящими инструментами. Не сомневаюсь, что есть альтернативные решения для каждой задачи, у каждого есть свои плюсы и минусы, и не всегда есть возможность говорить про то, что какое-то однозначно лучше другого, вопрос восприятия, я лишь привел свое, основанное на подходах Kafka Streams.
Значит мы друг друга не сможем понять. Спасибо!
По ttl тоже непонятно, я вам вроде ответил, что для cleanup.policy=compact нет ttl - перейдите по ссылке.
Ну вы то как раз и не ответили (до этого момента), а в статье по ссылке нет ни слова про ttl (буквально).
Как же не ответил, если написал про retention policy compacted и привёл ссылку и в самой статье и в комментарии, мне показалось что ответил, возможно не подробно, но по ссылке все нюансы описаны
Посылать читать статью вместо простого ответа на простой вопрос - это не ответил. В статье нет ни слова про ttl - мне погуглить проще было.
Коллега, я ни в коим случае никуда не посылаю, ваше право считать мои ответы неправильными. Я искренне хотел понять и ответить на вопросы предыдущего комментатора. Мне показалось, что я вполне себе ответил, дав ссылку на официальную доку, где можно почитать про разные clenup policy и в скользь упомянул про режиме compaction в статье, возможно не очень удачно, но я и не являюсь профессионалом в этом деле, я всего лишь любитель.
На мой взгляд при желании перейдя по ссылке можно ознакомиться со всеми нюансами, т.к. у человека видно, что нет пока необходимых знаний по этому вопросу, а вот то, что вы называете простым ответом повлекло ,бы за собой еще 10 простых вопросов, исхожу из того, что и в комментарии и в статье это объяснено так, что вы не поняли. Зачем я буду отвечать про ttl, если как вы правильно заметили такая терминология не используется в доках конфигурации - потом кто-то предъявит за ttl, что ввожу в заблуждение. Если есть действительно желание вникнуть в этот момент, то лучше идти к первоисточнику. Моей целью не было осветить все нюансы работы с Кафкой.
Так есть же режим у compacted топика для очищения данных по времени. Почему вы пишете что ttl нет?
Вы обладаете знаниями которых у меня нет. Выше все объяснено и приведены ссылки на оф сайт, там можно изучить как работает режим cleanup.policy=compacted. Да я видел, что был какой-то KIP, где можно задачать cleanup.policy=compacted,delete, но вряд ли вы про это
В документации и написано что compacted удаляет устаревшие записи с тем же ключём что у новых.
я этот момент как раз таки и объяснял в статье. в топике хранятся апдейты из БД, вычитав топик любой сервис может восстановить на своей стороне реплику БД, также подчеркнуто, что должен быть правильно выбран ключ.
исходя из вышеизложенного нас как раз таки и интересуют только последние данные по ключу, старые данные нас не интересуют. Для того, чтобы топик с одной стороны не разрастался бесконечно, так одной записи в БД может соотв несколько записей в топике, если мы будем ее обновлять, с другой - чтобы все данные по актуальным ключам не почистились как раз таки и выбрана cleanup.policy = compacted.
вас и предыдущего комментатора я не понял, думал вопрос по поводу что все записи удаляются по времени, это было бы проблемой для представленного подхода. То что старые данные ПО КЛЮЧУ удаляются, изза этого осознанно выбран режим compacted. Никаких проблем при этом не будет, реплика будет актуальна независимо от того, будет ли в топике по ключу несколько значений или только последнее.
данные по ключу в режиме compacted могут быть полностью удалены из топика, если послать recrod с value = tombstone.
Интересное решение, только не проще ли было бы проблему временной недоступности Postgre решать репликацией той части Postgre, которая должна быть всегда доступна. Всё-таки хранилище в Кафке - оно немного для других целей сделано
специфика работы Postgres в компании, что Postgres предоставляется команде как сервис и служба поддержки проекта не имеет доступа, такие требования были объявлены на старте проекта. Не исключаю, что можно сделать элегантней и проще, но из тех вводных, что были, показалось, что данное решение тоже достаточно простое и эффективное, тем более что в confluent guides вполне себе описаны схожие на мой взгляд кейсы.
В целом данная статья написана еще и из интереса того, как этот "велосипед" выглядит для не замыленного взгляда, так что не исключаю :)
Из преимуществ решения для меня это то, что я имею обновляемый кэш и в тоже время полную реплику справочника в топике.
Я не знаю, что там описано в Confluent Guides (не читал или читал настолько давно, что забыл про это), но Кафку можно изящным движением руки превратить в key value storage. А вот в хранилище, которое можно сканировать по диапазону ключей или вообще без фильтра, её лёгким движением руки не превратить - оно будет тупить
Не совсем понятно, что те, кто ставил вам такие требования, предлагал делать в случае, когда данные в момент недоступности надо не прочитать из базы, а записать в неё? Записывать на бумажку, чтобы потом через пару часов в базу перенести? Значение в справочнике может же быть и неправильным
Но в целом решение вполне рабочее, пока справочники - это что-то редко изменяемое
При желании при таком подходе можете реализовать такую структуру, которая будет индексировать по разному, насколько хватит вашей фантазии, но это да нетривиально. По умолчанию да согласен, Kafka это key value store и в kafka streams как раз таки на это упор, все реализации хранилищ строятся вокруг того, что это key-value store и даже в Rocks DB Java Api я не нашел ничего, кроме как put(key) get(key) remove(key), хотя в c++ api там есть и про диапазон тоже. Не каждое решение универсально, с этим спорить не буду. У нас кстати есть пару справочников, которые предоставляют api поиска по ключу и активным датам и по диапазону ключей и по префиксу, конечно же там не такая простая ConcurrentHashMap.
По требования возможно дал не так много вводных. Мне они кажутся вполне себе логичными. У нас есть Postgres предоставляемый как сервис. Сказано, что оооооочень редко может быть такое, что ночью нужно сделать какие-то работы, которые приведут к тому, что сервис будет лежать 2-3 часа. По нашему SLA мы не можем обрабатывать поступающую транзакцию более 10 мин ( т.е. нам несмотря на лежащую БД нужно обрабатывать транзакции).
"Редко изменяемые" довольно неточный термин, не знаю "редко у нас изменяемые справочники или нет" в вашей классификации, но как это выглядит: условно бизнес заключает договор с несколькими компаниями и несколькими банками, далее он передает службе поддержки, что к такому то числу нужно уметь обрабатывать поступающие транзакции, служба поддержки готовит справочные данные к данной дате, обычно проводится какое-то боевое тестирование в проде на столе. Таких запросов от бизнеса может быть несколько в неделю, что выливается в изменение/добавление/удаление сотен записей в нескольких справочниках. Может быть и больше в зависимости от нагрузки от бизнеса
Данный подход вполне себе покрывает все данные требования, изменения в справочниках быстро доставляются до сервисов с помощью представленного механизма. В тех исключительных \ситуациях, когда постгрес лежит ночью справочники понятное дело никто и не думает заполнять, все эти работы проводятся днем, в ночное время работают только дежурные.
По умолчанию Кафка - это журнал сообщений, и я знаю довольно много людей, которые использовать её даже в роли key value storage считают извращением. И я их в чём-то понимаю - это уже где-то довольно близко к забиванию гвоздей микроскопами, она оптимизирована совсем не под это - хотя да, может. Почему Кафка не предоставляет базу с произвольным доступом - потому что там это key-value store устроено как таблица ключей со ссылками на оффсет в логе (ну либо как честный key value store на консумере из очереди как второй вариант - типа вашего решения, но не в памяти). Любой скан не по ключу - это скан этой таблицы с лукапами по всему журналу вразброс, это просто не может работать быстро, хоть убейся, by design
Ещё раз - я не критикую ваше решение и искренне готов вас похвалить (абсолютно безо всякой снисходительности и иронии) за креативность. Мне бы такой вариант наверное в голову не пришёл, я бы сделал что-то более унылое и требующее дополнительного софта. То есть то, что ваше решение мне кажется странным - это скорее моя проблема, чем недостаток решения
Если справочник будет часто изменяемым (например, в справочнике будет волшебное поле "последняя цена", которое меняется после каждой транзакции, видал я такие справочники) - в Кафке постоянно будет что-нибудь новенькое и вся эта схема начнёт выдавать интересные спецэффекты, я это имел в виду. Но пока такого нет - всё должно быть хорошо
спасибо за оценку!
"По умолчанию Кафка - это журнал сообщений, и я знаю довольно много людей, которые использовать её даже в роли key value storage считают извращением. " - кажется что ваши знакомые не пользуются всеми возможностями кафки, т.к. confluent в целом пропагандирует другое и много докладов в том числе про это, даже есть такая ksqlDB - где DB прям в названии, хотя конечно это не совсем про DB :)
Я с вами полностью согласен, что представленное мной решение очень специфической для кейса с множеством предусловий и оговорок. В целом такая идея возникла после изучения примеров на сайте confluent по построению kafka stream application и прочтения Kafka Streams In Action. Даже думал начать использовать kafka streams. Но во первых понял, что у нас уже 10ки сервисов написаны с использованием реактивного клиента, которые просто так не перепишешь и кажется, что следуя рекомендациям этих гайдов эти сервисы надо немного по другому организовывать и разбивать на сервисы. Также мне на самом деле не очень понравилось то, как написан код как раз таки с этими Store, из-за иерархии наследования, которая была там, очень сложно кастомизировать так, как мне хотелось. Поэтому пока отложил этот и решил для начала взять оттуда принцип для одного специфичного кейса со справочниками. Т.е. сам kafka streams framework на первый взгляд субъективно не понравился, но с точки зрения описания принципов построения архитектуры приложения на базе Kafka - очень вдохновило, так что сейчас от этого активно использую kafka connect и вот представленное решение.
Я уже упомянул в другом комментарии, что изначально на этапе mvp эти справочники вообще велись в Redis. Но Redis это:
1) это дополнительная точка отказа ( сам кластер надо содержать и поддержать, у нас кстати на проде пару раз была потеря кластера)
2) это дополнительное межсетевое взаимодействие, на которое тратится время при обработке транзакции
3) неудобство ведения ( можно конечно пофиксить это с помощью такой же репликации из postgres через кафку)
А что касается справочников, хотя все таки это наверно не совсем справочники, где что-то динамично меняется, в зависимости от обработанных транзакций например - на этот счет можно также найти примеры в вышеупомянутых ресурсах на confluent, конкретно представленное решение в таком виде не подходит, согласен.
Они знают про ksql DB (мы используем ksql DB в production), отчасти потому и не в восторге. Хотя я лично скорее одобряю. Но там конечно есть свои подводные камни, в первую очередь про то, что оно не SQL, как бы ни притворялся, и самое больное место - как всю эту машинерию надёжно и безболезенно деплоить
А как решался вопрос при горизонтальном масштабировании сервиса, в память которого вычитывался справочник? т е если доп инстанс сервиса поднимется, как он сможет получить справочник в памяти?
точно также, любой инстанс будет работать одинаков - загружать всю реплику из топика полностью. В пункте "Kafka Store Component" я об этом написал, что есть в kafka streams партицированные хранилища, есть Global Store, вот нам пока достаточно концепции Global Store - когда мы полностью всегда при старте приложения вычитываем все данные топика из всех партиций, а затем уже приложение начинает делать свою остальную работу используя справочник, при этом справочник продолжает обновляться.
Т.е. если мы запустим 10 или 20 инстансов, все 10 или 20 инстансов каждый вычитают весь топик и будут содержать в себе полные реплики справочников. Собственно про это код KafkaGlobalStore.Т.к. это Global Store можете масштабировать сколько угодно без специальных настроек и оглядки на количество партиций в топике-справочнике, все зависит только от остальной логики вашего приложения. Данная реализация никак вас не ограничивает в масштабировании.
в проекте не используется распределенный кэш(редиса например)? кажется, что подобная задача с распределенным кэшем решается более естественно, при этом пропадает необходимость хранения снэпшотов справочников(их ведь может быть много) в памяти каждого экземпляра сервиса.
о, да был у нас и даже еще где-то остался кластер редис:)
конечно есть разные подходы и решения, но от redis как раз в какой-то момент отказались
Давайте порассуждаем, уточните, что вы имеете ввиду под распределенным?
Вы имеете ввиду, то, что в каждом инстансе мы запускаем внутри ноду редиса и все их объединяем в кластер? или вы про централизованный кластер редиса?
Данные в редис вести неудобно - это опробовано, служба поддержи писали скрипты на питоне для загрузки и обновления 1000 терминалов, хотят писать sql запросы, т.е. их все равно нужно как-то туда реплицировать из реляционной БД.
Реализация хранилищ ограничена структурами редиса - с этим тоже сталкивались, не хватало.
Представленное решение для работы со справочниками не использует никакого межсетевого взаимодействия, кроме загрузки данных на старте и фонового обновления, что тоже не так плохо. Оба варианта развертывания редис все равно предполагают межсетевое взаимодействие и влияют на скорость обработки.
Опять же не претендую на то, что это самое оптимальное и правильное решение. Возможно я не дал всех вводных, возможно есть решения лучше. Но конкретно с Redis пока не вижу как это можно сделать проще и эффективней, не исключаю что можно, но сейчас мы таки отказались от Redis.
Справочники существовали и до этой конструкции с Кафкой? Если да, то как происходило изначальное наполнение Кафки данными из справочников?
Ещё вопрос, связанный с предыдущим. Как я понял, данные в Кафке у вас хранятся в памяти. Что будет, если кластер Кафки полностью перезапустится по некоторой причине?
Как я понял из ответов выше при бутстрапе сервиса происходит чтение всех топиков Кафки и последовательное воссоздание хранилища заново в оперативную память.
"данные в кафке хранятся в памяти" - не понял утверждение. Возможно вы имели ввиду данные Из Кафки хранятся в оперативной памяти сервиса - тогда ДА.
Что будет если кластер Кафки перезапуститься?
Если Кластер Кафка не работает - то вся наша система ничего не может делать - это сказано в предусловиях, так что если понадобиться перезапуск Кластера Кафки, то мы скорей всего просто остановим все сервисы, разберемся с кластером, перезапустим его и перезапустим сервисы. Если перезапуск произойдет внештатно, то мы не сможем обрабатывать транзакции дальше до восстановления. Смогут ли сервисы восстановить работу после перезапуска кластера без перезапуска самих сервисов - надо тестировать каждый конкретный сервис и то, как там реализована работа с консюмерами.
Но повторюсь, Кафка у нас считается высокодоступной, поэтому при таком редком маловероятном кейсе легче все выключить и включить, чем тестировать и поддерживать работу при таком кейсе. Поэтому вдруг для вас это важно, переподключение консюмера при перезапуске KafkaGlobalStore, то возможно его нужно немного доработать. Для нас такое усложнение логики не актуально.
А не сравнивали сколько занимает инициализации кеша напрямую из базы, приведены метрика только для инициализации из Kafka - 2.5s
Нет не сравнивал, т.к. не особо интересно. Убежден, что такую выгрузку можно сделать быстрее, но для меня это не так важно. Мне нужно было убедиться, что на кратно большем объеме инициализация справочника выполняется за удовлетворительное время, т.е. время не сильно отличимое на фоне запуска spring boot сервиса который работает с кафкой ( в самом лучше случае это тоже занимает секунду, но иногда и больше, т.к. требуется время на то, пока консьмерам назначаться партиции, пока произойдет ребалансировка - это может занять и 10ки секунд).
Если я вас правильно понял, вы хотели предложить подход, при котором загружать данные в кэш из БД напрямую а потом периодический его обновлять. Если так, то такое решение не подходи, т.к. если сервис начнет перезапускаться в момент, когда БД не подключена - он не сможет подняться.
ConcurrentHashMap - при описанной архитектуре избыточен. Достаточно обычного HashMap.
У вас же параллельный доступ к этой структуре только на чтение.
Модификация, а точнее создание происходит изолированно в отдельном потоке.
у меня сомнение на этот счет. hash map не tread safe. любое добавление элемента в hash map может привести к перестройке внутренней структуры, таким образом не уверен, что не будет спецэффектов при чтении из других потоков. возможно вы что-то лучше меня знаете или понимаете и в кейсе когда только один поток пишет, а другие читают должно работать или я пропустил какие-то изменения в новых версиях jdk.
В любом случае даже если так, я обычно следую принципу: если есть сомнения - используй thread safe классы, такие как Atomic*, ConcurrentHashMap и др. По-моему на каком-то очень крутом докладе на joker, где разбиралось эффекты влечет объявление volatile и какие можно не помечать volatile, в конце был дан совет - при сомнениях лучше использовать оптимизированные под это спец классы, такие как Atomic* ConcurrentHashMap и др, потому что даже зная все детали того, как компилируется и оптимизируется код в jvm, можно не туда свернуть в рассуждения и прийти к неправильному выводу. Так что если мне нужна HashMap в многопоточное приложение, то я даже если честно долго не думаю и беру CHM.
Кажется, что даже если вы правы и я что-то упускаю и недопонимаю, вряд ли выигрыш от использования HashMap будет вообще различим в такого рода сервисах, а вот риск получить спецэффекты при работе с HashMap, когда по логам значение будет записано, а в итоге не будет вычитано, точно не хочется потом разбираться с этим.
один поток пишет, а другие читают
Обычно изменения не единичны и чтобы потребители видели консистентные данные делают сначала заполнение данных в структурах в памяти полностью и только потом эти структуры становятся доступны на чтение.
Самый простой вариант через подход с двойным буфером.
Но выбор конечной структуры зависит от типа данных, какие модификации и как часто, какие требования к консистентности и к потребляемой памяти.
вот риск получить спецэффекты при работе с HashMap
Само использование ConcurrentHashMap не отменяет всех возможных спец эффектов, но дает ложную надежду, что их не будет.
Я если честно вас не понимаю, особенно про 'ложную надежду' в chm. Данные именно что обновляются в рантайме и перестройка структуры может произойти в любой момент. Моё понимание java memory model и ваши обьяснения на данный момент не позволяют с вами согласиться. Повторюсь chm или hm в рамках моих сервисом будут не различимы, но предпочитаю использовать при малейшем сомнении thread safe структуру. Не важно могут ли быть спецэффекты которые я могу предположить или такие, которые я не могу. Я вот честно не хочу даже об этом думать, есть thread safe структура из стандартной библиотеки и если с ней будут спецэффекты, то я по крайней мере буду уверен, что это не из-за того, что я использовал не thread safe структуру.
Если вам нужно изменить десятки записей в CHM, то атомарными у вас будут только "точечные" изменения. Другими словами пока вы потокобезопасно ее правите у вас потребители этой структуры читают неконсистентные данные.
Другими словами - происходит так называемое грязное чтение.
Одно из решений этого - использовать подход с двойным буфером. А это означает, что CHM никакой пользы не принесёт.
меня устраивает то, что вы описали, у меня нет цели, чтобы мгновенно получить точечные изменения, в целом сама доставка от момента добавления записи в постгрес, добавления в топик и вычитки занимает тоже довольно неопределенное время и мы к этому готовы, выполняем конфигурацию заранее. Зато я не думаю о том, как устроена внутрянка HM и какие эффекты могут быть в многопоточном приложении.
я повторюсь, я не хочу думать и фантазировать об эффектах и рисковать, мне просто не нужна такая оптимизация, использовать HM или CHM для меня скорей всего при тестировании будет даже статистически неразличимым. Главная оптимизация уже сделана, за данными не ходим в редис, а обращаемся к In-Memory структуре.
Спасибо, что поделились предложением, но меня не убедили. Ваше право так подходить к оптимизации ваших приложение, мне это кажется излишним, даже если это действительно правда.
Речь не про времени доставки записи из пострела до кэша в памяти. Речь про то, что когда справочники развесистые и записей прилично. То их обновление в памяти занимает время. Если в момент обновления кто-то одновременно читает оттуда данные и исполняет на осное этого бизнес-логику - могут быть проблемы, т.к. данные в кэше неконсистентны (грязное чтение). И отлавливать ошибки из-за этого очень нетривиально.
и CHM структура не помогает решить эту проблему. CHM про безопасность атомарных операций.
Поэтому говоря, что CHM здесь не нужен я не имею ввиду, что он будет тормозить на столько, что его нужно заменять. Но с CHM может быть необоснованная уверенность, что у вас все хорошо с асинхронкой когда на самом деле это не так.
Kafka как хранилище справочников