Алексей Кашин — Надежно отправляем события в Apache Kaf… — Transcript

Доклад о надёжной отправке событий в Apache Kafka с использованием CDC и паттерна Transactional Outbox.

Key Takeaways

Гарантированная доставка сообщений в Kafka требует компромиссов между скоростью и надёжностью.
Паттерн Transactional Outbox помогает решить проблему консистентности между базой данных и Kafka.
Распределённые транзакции и паттерн Сага — основные подходы к решению проблем консистентности в распределённых системах.
Правильная настройка продюсера и обработка исключений критичны для надёжной работы с Kafka.
Понимание работы vacuum в Postgres важно для поддержания производительности базы данных.

Summary

Обзор проблем доставки сообщений в распределённых системах и микросервисах.
Объяснение уровней гарантии доставки сообщений в Apache Kafka: от максимальной производительности до exactly once.
Рассмотрение особенностей работы с базой данных Postgres, включая процесс vacuum и его виды.
Обсуждение типичных проблем при отправке событий в Kafka и вопросы консистентности данных.
Описание решений для обеспечения консистентности: распределённые транзакции, паттерн Сага и архитектурные изменения.
Введение в паттерн Transactional Outbox как надёжный способ отправки событий.
Технические детали настройки продюсера Kafka и обработки транзакций.
Практические советы по работе с конкуренцией и хранению смещений в Kafka.
Обсуждение ограничений и нюансов реализации паттерна Transactional Outbox в реальных проектах.
Рассмотрение вопросов поддержки базы данных и оптимизации работы с Kafka.

Chapters

Full Transcript — Download SRT & Markdown

Speaker A

[музыка] Я рад вас приветствовать на нашем докладе, посвящённом надёжной отправке событий в Apache Kafka. Но прежде чем мы приступим к докладу, ответьте мне, пожалуйста, на пару моих вопросов. А кто, поднимите руку, пожалуйста, те, кто сталкивался с задачей отправки сообщений

Speaker A

между несколькими системами? Отлично. А поднимите ещё руку, пожалуйста, те, кто терял сообщение как-либо когда-либо, возможно, из-за несогласованности транзакций, каких-то проблем при отправке. Отлично. Да, это очень большая проблема, которая нам всем знакома. Вот. Меня зовут Владимир Котяев, я из банка. Теперь уже рядом со мной мой

Speaker A

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

Speaker A

pa of Link. Но сегодня будем чуть подробней и более детально рассказывать про CDC и pattern transactional outbox. Вот я уверен, что доклад будет полезен и для тех, кто уже практикующий инженер, и кто только начинает своё приключение в этот

Speaker A

прекрасный мир гарантированной доставки сообщений. Ну а я передаю слово Алексею, который нам сегодня всё расскажет. Спасибо. Спасибо, Вова. Давайте потихоньку начинать. Видно презу? Самый бесполезный слайд сегодняшней презентации. Надеюсь. Меня зовут Алексей, я работаю в ТАБАН. Погнали. В общем, чтобы не говорить

Speaker A

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

Speaker A

читают у нас данные. У кого так? Вот. Давайте вот сейчас холодно, руки не стесняемся поднимать, чтобы потеплее было. Отлично. Но это не до конца рабочее решение, естественно. А системы хотят узнавать: а что же у нас когда-нибудь поменялось? Кто-то поме... од хочет

Speaker A

запускать какие-то там, знаю, процессы, заявки, что-то ещё. Не так важно. Как такую проблему решать? Естественно, мы ставим посередине очередь и начинаем в неё писать. Да, мы отправляем данные, потребители получают эти события и реализуют свою логику. Всё стандартно, всё

Speaker A

понятно, паттерн известный. Вот прежде чем пойти дальше, я хочу немножко затронуть технологию, на которую в качестве основного хранилища так круто почти все. А кто использует Кафку в качестве брокера стриминга? Супер, всё знаете, наверное. А во-первых, у Кафки есть такая штука как

Speaker A

гарантии доставки. Это то, как наш продюсер доставляет сообщение до брокера. Гарантия доставки — это такой трейд-офф между скоростью работы наших систем и надёжностью того, как мы данные доставляем до брокера. Первый уровень — когда нам нужна максимальная производительность, и поэтому мы можем

Speaker A

допускать какие-то недоставки. Да, понимаем, что что-то будет не доставлено, продюсер наш он не дожидается подтверждения от брокера, когда отправляется событие. Я надеюсь, никто так не использует. Да? Кто использует? Кто-то есть? Один. Вот. Средний уровень, наверное, самый частый, самый

Speaker A

популярный — это сообщение мы доставляем до брокера гарантировано хотя бы один раз, но при этом возможны повторы. Дожидается подтверждение, но мы знаем, что системы у нас распределённые, что-то может потеряться, событие записалось, но на самом деле подтверждение не дошло. Поэтому нам нужно сделать какой-то

Speaker A

ретрай. Вот средний уровень. И самый такой жёсткий, самый максимальный — это exactly once. Это уже такая сложная настройка продюсера, консьюмера, импотент-продюсеры. Нужно понимать, что они тоже работают на экземплярах, они во всём кластере, поэтому тоже возможно дубль сообщений. Мы используем транзакции

Speaker A

Kafka, чтобы гарантировать доставку до Кафки. Вот. А сложно, и даже технически на всех уровнях возможны проблемы. Вот поэтому такой уровень используется довольно редко. Сразу говорю, наш выбор — это, естественно, at least once. Мы можем допускать дубли событий, не только, эээ,

Speaker A

если нам какая-то такая нужна, просто не доставка. Да, но мы можем по каким-то своим процессам действительно переотправить данные, просто договоримся с потребителями, всех всё устраивает. Вот. Если хотим узнать про вот эти вот уровни доставки, то есть прекрасный доклад Виктора Гамова Q код,

Speaker A

ссылка, переходите, смотрите, там очень круто всё расписано. Ещё важная тема при работе с ПО. Скажите, а кто знает, что такое вакуум? Вот используют Postgres много, а про вакуум знают не так много. Давайте поправим. Работает по MVCC, много версионная

Speaker A

архитектура. Что это значит? Когда что-то меняем в текущей нашей строке, на самом деле мы не в ней меняем, мы создаём новую её версию, а старая версия сохраняется. Вот вакуум он нужен для того, чтобы наша база не росла всегда чрезмерно. Да, это

Speaker A

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

Speaker A

диске, то есть таблица как весила у нас какой-то определённый объём, vacuum отработал, она столько же весит, но там данных как бы чуть меньше становится. С другой стороны, есть vacuum full — это уже процесс обслуживания базы данных, он нужен для того, чтобы действительно

Speaker A

сжимать таблицы. Он блокирует таблицу, и поэтому выполнять его на проде в рантайме, в рабочее время, довольно опасно, но иногда нужно. Вот. Ну и есть автовакам — это фоновый процесс, который периодически запускается, удаляет эти мёртвые строки, почищает команду vacuum и команду analyze.

Speaker A

Вот и vacuum, автовакам нам помогает поддерживать базу в хорошем состоянии без необходимости вручную запускать vacuum. Такая эвристика. Вот. А естественно, об этом есть документация. Но если хотим поглубже, есть несколько статей от pog pro, ребята очень хорошо расписали, как

Speaker A

работают очистки, автоочистка есть, потом переходите, читайте. Вот. Но вернёмся к проблеме. Да, мы хотим данные отправлять в очередь, кто-то будет из неё читать. Самое простое решение, какое будет? Да, просто поставим после сохранения в репозиторий отправку в очередь через стандартный Kafka телей.

Speaker A

Все же такое делали? Да? Правильно. Но если мы, разработчики, уже чуть попытнее, мы понимаем, что тут возможны проблемы, и нам нужно ответить на несколько вопросов. Во-первых, а нужно ли нам сохранять, ну отправлять данные, вернее, сохранять в базу, если Kafka

Speaker A

недоступна? Да, как тогда быть? Может быть ретрай, может быть отложить, может быть вернуть пользователю ошибку. Да, нужно на это просто ответить. Что делать, если наоборот база недоступна, а Kafka доступна? Нужно ли нам отправлять события? Обычно нет, но в каких-то кейсах возможно

Speaker A

и нужно. И ещё такой интересный вопрос: а как нам реагировать на задержки? То есть мы понимаем, что наши системы работают не всегда стабильно, мы можем отправлять события, но они просто могут доходить медленно. Как нам тогда жить? И

Speaker A

вообще, является ли это проблемой? Что только что писал — это стандартные такие проблемы консистентности в распределённых системах. Да, у нас Kafka не является хранилищем, но в данном случае выступает как источник, вернее как приёмник, да? Как решается? Если мы сейчас зайдём в Google

Speaker A

и спросим, как решать проблему консистентности в распределённых системах, или в ChatGPT спросим, что сейчас моднее, нам выдаст три решения: это распределённые транзакции, использование паттерна Сага либо переделка архитектуры под SCS инсорсинг. Ну последние — это прямо реальная переделка архитектуры,

Speaker A

поэтому я не буду рассматривать. А вот первые два чуть-чуть расскажу. А во-первых, распределённая транзакция решается, там используется двухфазный коммит, ну или трёхфазный коммит. Что такое? Наш сервис, которым мы сохраняем все события, он выступает в роли координатора. Он сохраняет данные в базу данных, но не

Speaker A

коммитит их. Да, он отправляет данные в Кафку, и если получил ошибку, он делает лбк из всех систем. Вот решение, на самом деле, не сильно простое, хоть выглядит довольно просто. С другой стороны, Сага. Сага основана на компенсационных действиях. Мы сохраняем события в базу

Speaker A

данных, коммитим их, отправляем события в очередь, получаем ошибку, мы должны придумать компенсационное действие, которое, ну либо удалит нам ту эту строку, либо вернёт старую её версию обратно, но в э...

Speaker A

консистентной состоянии то есть у нас получается такая cons она когда-нибудь дойдёт до консистентной у нас консистентная система или пользователь получат некон систентки вот задержки довольно интересная тема кто использует Tom и стандартный Немного это хорошо но вы наверное знаете кто использует Что или

Speaker A

кто не использует использует по такой же принесли за транзакции наши то у нас коннекты встанут в очередь и поэтому сервис на замедлится очень сильно А может даже упасть тоже нужно об этом помнить и Неужели нет решений сейчас на

Speaker A

рынке которые нам вот такое решают И вообще когда мы говорим о стриминг данных с использованием каких-то баз данных сразу же вспоминаем если кто че cdc решают определённые проблемы для чего вообще такие решения создавались например нам нужно выводить какие-то медленные данные в другую базу

Speaker A

для аналитики часто ведь такое делаем да чтобы на проде не селекти Вот либо нам нужна какая-то идентичная копия Ну короткая обычно для восстановления не устраивать стандартный КП нам нужно там последние 15 минут хранить чтобы быстро восстановить да либо нам Нужно

Speaker A

настраивать etl процес extract transform и тогда мы просто делаем такую поставку данных вот такие проблемы решают cdc решения как они их решают А есть два подхода либо Мы периодически данные выбираем и отправляем куда-то Либо мы постоянно следим за изменениями и транслируем их в

Speaker A

реальном времени как мы это делаем Ну либо запросами пишем селекты отправляем либо используем механизмы триггеров которые есть в любой СУБД и на них отправляем либо самый сейчас модный самый популярный инструмент это использование журналов транзакций Лога вала для того чтобы там

Speaker A

находить изменённые данные и отправлять уже их тогда мы базу данных не нагружая такие решения не нагружают базу данных а примерами cdc решений обычно сейчас когда говорят о cdc все подразумевают deum да ведь но это не единственное решение deum действительно мощный

Speaker A

фреймворк открытый для cdc ну как бы он работает на разных базах данных mygo э стримит в разные места всё хорошо если работали с оралом скорее всего слышали про Oracle Golden Gate Очень мощное быстрое решение но при этом очень

Speaker A

дорогое и сейчас к сожалению для нас оно недоступно вот ну и каждой Cloud платформы вот я выписал aws Database migration Service у каждой Database у каждой Cloud платформы есть такие решения то есть их довольно много не только deum является cdc

Speaker A

А так как это коробка Да мы уже применительно к проблеме то у коробки есть всегда ряд каких-то недостатков потому что коробочные решения они создаются для общего класса проблем для общего класса задач но они всегда подходят под нашу конкретную и могут под

Speaker A

неё не подходить например А кто работает с авра кто стримит данные в кафку в авра Вот немного Те кто не стримит Задумайтесь об этом Если не знаете посмотрите что это такое когда мы работаем с фреймворка типа там сложно настраивать именно кастомные

Speaker A

авра схемы он генерирует авра схемы по таблице вот мы работаем с такой таблицей с такими полями вот такие вот авра схемы будут сгенерировано опять же так это коробка мы не можем управлять Э задержками да нет гарантии своевременной доставки всё идёт хорошо всё нормально

Speaker A

что-то случилось Не то мы не знаем Как править Ну перезапустили вроде заработало как проблему решать Ну там поддержка всё такое а и опять же если мы делаем сущности Наши в нашем сервисе на нескольких таблицах то cdc решения будут

Speaker A

будет сложно настроить именно отправку всей сущности потому что работают о не как я говорил по таблице изменения в одной таблице прошли он её отправил А то что находится в других таблицах решени не знает Вот про тоже был прекрасный доклад на

Speaker A

одном из предыдущих конференций переходите смотри по какой-либо причине по одной из моной озвученных или по какой-то другой не подошло Мы хотим написать что-то своё тогда стоит обратиться вот к такому паттерну Trans первый раз я о нём узнал на бумаге

Speaker A

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

Speaker A

таблицу а затем У нас есть отдельный компонент А может отдельный сервис который выбирает эти события уже отложено и отправляет их там же можно и ретрай сделать и всю безопасность подкрутить всё что нам нужно вот после успешной отправки событие оно удаляется

Speaker A

ЛЕКО забывается ом Давайте попробуем его реализовать Для этого нам нужна по крайней мере outbox таблица Как называется неважно нам важно поле payot вот Аде курсор вот поле P куда мы будем сохранять А наши события для отправки основные остальные поля тоже нужны сейчас к ним

Speaker A

вернёмся сразу помним чуть раньше говорил про вакуум Да и то что события после отправки нам не нужны нам нужно их удалять если мы будем удалять мы будем триггерить этот механизм mvcc который будет триггерить вакуум Как удалять правильно в постгрес правильно

Speaker A

удаляется с помощью партиции партиции по сути это отдельные таблицы в базе данных которые сбд считает одной табли для польва на как разработчиков это ВС прозрачно мы делаем за просток мастер табли А нам уже выбираются данные из нужных партий партиции удобно создавать

Speaker A

по дате Вот как я тут сделал в таком случае первичным ключом будет не просто идентификатор но ещё и ключ партиции да То есть дата К сожалению по из коробки просто так не умет создавать партиции и удах этом знает обя

Speaker A

научить нашу базу данных так работать есть timescale db кто использует Times db во вы крутые это расширение для поса там есть объект гипер таблиц можно настроить такое декларативное удаление Ну или создание удаления если нет Ну тогда нам нужно либо какой-то Рон Job

Speaker A

периодически запускаемый который будет создавать партиции и удалять их да либо внутри приложения тоже напишем какой-то код Вот пример создание создание да Да и чуть ниже удаления стандартными ddl операциями Да спарти сями разобрались А как записывать в нашу таблицу тут не буду

Speaker A

каких-то конкретных примеров приводить все инженеры все такие задачи решают Можно либо вручную можно использовать аоп либо прокси механизмы фреймворка либо можно использовать события orm которые мы используем потому что не всегда пишем базу данных через gbc Тей какой-нибудь условный

Speaker A

Да ну вроде записали Как отправлять абстрактная реализация отправки она простая выбираем и в цикле отправляем Мы же не хотим по одному событию скорее всего Да мы хотим это делать пачкой выбираем несколько событий и в цикле отправляем всё просто Вот Какие тут

Speaker A

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

Speaker A

повторных отправок они возможно по бизнес логике Но специально несколько раз отправлять наверное будет Неправильно и нельзя допускать пропусков событий при отправка Давайте поговорим про отправку данных и про то как данные нам из або таблицы выбирать Какие есть способы

Speaker A

потому что их тоже оказывается несколько когда мы говорим про выбор чего-то из базы да В таком формате Мы всегда думаем что ну это очередь как очере реализуется на реляционных базах через механизм пессимистический блокировок SY update тогда мы в транзакции выбираем

Speaker A

эти строки другой ДБ запускается и он просто видит что эти строки заблокированы либо встаёт на них либо если у нас есть такая инструкция как Skip он пропускает это выбирает следующее очередь красиво вроде нам подходит выглядит это дело уже в нашем

Speaker A

коде что нам важно здесь важна транзакционная нам нужно поставить аннотацию transactional чтобы всё событие весь всё время отправки у нас была транзакция тогда эти строки останутся заблокированными никто их не возьмёт ещё раз и не отправит их есть ли в этом коде

Speaker A

проблемы как думаете Ну вообще есть долги транзакции а так как у нас механизм отправки он может какое-то время выполняться а возможное задержки Да и выполняться будет очень долго они влияют на производительность СУБД э вся блокировка строк она повышает нагрузку

Speaker A

на базу данных если строк будет очень много заблокировано долгое время суд вообще может заблокировать нашу таблицу и тогда мы даже записать в неё не сможем это замедлит всё наше приложение вот ну с этим тоже можно жить можно побороть

Speaker A

Давайте введём статусы а нам нужно поле статус Вот здесь Да какой-то ворр мы подразумеваем что там будет три статуса строка Новая строка заблокирована Ну то есть какая-то обработка ведётся и строка отправлена да нам нужен индекс обязательно чтобы искать только новые строки мы вот такой

Speaker A

индекс повесим на статус New тогда строки будут выбираться быстро и Запрос который будет нам в одной транзакции апдейт строки ставить им L если у них ещё New Ну какое-то ограниченное количество Наверное нужно сделать а как это происходит здесь уже

Speaker A

транзакций нет Да тут вот сверху ничего нету нам здесь не нужна транзакция на всё это будет в разных транзакциях скать Сначала мы блокируем записи выбираем их отправляем и потом делаем ставим маркировку что они отправлены мы полечить что А у нас могут зависнуть события в

Speaker A

статусе LED Да ну то есть прошёл момент что мы заблокировали события отправляем что-то упало сервис наш упал события остались вть статусы нужен какой-нибудь отдельный компонент Джок чтобы эти события подбирать и переотправить Как можно сделать можно выбирать по сдвигу либо по смещению Да мы В

Speaker A

отдельной таблице будем хранить информацию о последние отправки нам нужна отдельная таблица я назвал её Out BX Counter У нас есть какое-то значение нашего каунтер пока неважно Какое чуть позже дойдём А как будет Вот у нас есть временной ряд

Speaker A

У нас есть какие-то события у нас запускается наш Джоб он отправляет заснул отправляет дальше Да Всё хорошо Всё идёт Так примерно выглядит Вот Нам нужен объект какого-то кантера Да мы его инкремент кантера и отправляем их точно также в цикле а в конце апм каунтер на

Speaker A

новое значение то есть уже Мы старые не возьмём когда мы говорим о кантере да мы сразу таки думаем Ну есть же у нас идентификатор автоинкремент удобно sequence bal можно использовать его Да можно просто ставить границы и всё будет прекрасно

Speaker A

работать наверное но есть вот такая проблема когда мы используем идентификатор это ставка с меньшим идентификатором строки на самом деле в нашу таблицу попадают в разном порядке это может случаться из-за того что у нас сервисы умные и они каширу

Speaker A

идентификаторы внутри даже если это если мы это победим то возможны вот такие ситуации транзакции У нас тоже не моментально зака они могут немножко задержаться по какой-то причине сетевой база данных неважно какой мы начинаем транзакцию Да с вставкой идентификатора

Speaker A

ой как-то мышка не очень двигается с идентификатора 100 она почему-то подвис Да мы уже вставили 1001 и 102 и потом только вставили 100 если мы в это время запустим же отправки он выберет данные по 102 отправит их то наш вот этот вот

Speaker A

несчастное событие с идентификатором 100 оно потеряется вот как-то такое тоже нужно решать как решать если мы хотим остаться на идентификатора Ну каждое решение оно уже будет бить по производительности нашего сервиса нам нужно либо поднимать уровень изоляции до максимального

Speaker A

максималь у нас поточно либо нам нужно как-то придумывать механизм который будет блокироваться или синхронизироваться на вставке оно будет быстрее чем уровень изоляции Да поднять но всё равно не сильно быст тоже возможны проблемы с другой стороны мы можем использовать не идентификатор

Speaker A

Как число А можем использовать время у нас тем более есть партиции по времени да можем просто запоминать время до которого мы отправили а затем искать уже с него тогда вот если мы сделаем такую определённую задержку интервала Да вот

Speaker A

прям в Запроси можем узами ставится чуть по какой-то причине чуть позже и проблемы не будет а но тут нужно договариваться с потребителями что у нас будут события доходить С какой-то задержкой если и МОК это нормально но при этом ещё есть ряд проблем все

Speaker A

понимаем что работаем Мы в распределённой среде у нас сервисы находятся в разных местах они могут быть географически распределены и на них может быть вообще разное время да А кто может Наверно кто сталкивался с проблемой что периодически у нас системы

Speaker A

подстраивает время кто-то знаком с такого вот прекрасно это тоже нужно помнить и тоже учитывать когда мы вот выставляем такой интервал для отправки Ну вроде как проговорили как нам данные выбирать Да А давайте поговорим про то как их правильно

Speaker A

отправлять потому что тоже возможны как оказывается варианты код синтетический э Кто думает что в нём есть проблема почему-то немного людей думают что в нём есть проблема а она есть и она этон Да И вот если посмотреть вот что у

Speaker A

меня здесь написано н Кате Да блин очень сложно это сделать на самом деле цент он не отправляет событий в брокер он отправляет событие в специальный буфер который называется Record accum это часть библиотеки кавка продюсера все фреймворки все какие-то

Speaker A

стартеры которые мы используем они скорее всего используют е почему кака такая быстрая когда мы отправляем события они накапливаются в Буре отправляются пачкой если бы мы каждое событие отправляли это было бы очень долго буфер настраивается у него есть ряд

Speaker A

параметров Первый параметр MS это как Долго продюсер ждёт ответа ждёт отправки пакета в миллисекундах мы настраиваем это время время подошло события отправились дальше Проходит время снова события отправляется есть другой параметр размер пакета в байтах не штуках наши коллеги на этом столкнулись

Speaker A

один раз это размер пакета в байтах который будет отправлен даже если события ещё даже если linger MS время не наступило Вот и третий параметр buffer Memory максимальный размер буфера если буфер переполниться то тогда СН будет блокироваться до отправки всего буфера

Speaker A

для чего это нужно Ну чтобы мы случайно не словили оо если у нас очень много событий будет генерироваться для отправки мы можем накопить столько что у нас память закончи поэтому механизмы каки они умные они нам помогают в этом

Speaker A

деле вот если мы посмотрим на сигнатуру меда И вообще когда мы говорим про что-то что случится в будущем и в Джаве Это какой-то скорее всего объект Да если мы посмотрим на он возвращает нам объект вот объект уго нужно дождаться когда эти колбеки

Speaker A

сработают там есть колк на успешную отправку и колк на неуспешных бы гарантировать отправку мы должны на эти колбеки правильно уметь обрабатывать настраивать Трай возможно что-то ещё делать но если мы будем делать Так по одному просто в цикле то это будет

Speaker A

медленно Почему медленно потому что мы отправим событие в буфер и буфер будет ждать пока пройдёт н MS вот он прошёл событие отправилось соответственно тут мы можем вобще словить ситуацию Когда у нас события поступают гораздо быстрее чем мы можем их отправлять вот можно

Speaker A

делать пачкой Ну по крайней мере то есть аккумулировать Ну я вот привёл пример через F аккумулировать события в и уже бло на ней часть событий уже отправится часть за отправки это будет быстрее вот ну либо вариант из использовать ка

Speaker A

транзакции тоже нетривиальный механизм нам нужно правильно настроить продюсера А коммитить транзакцию и обрабатывать очень много исключений которые вот здесь у нас есть которые возможны и правильно их обрабатывать но тоже возможно соответственно главное помнить что нужно обрабатывать то что мы в кафку

Speaker A

отправляем А если хотим поподробнее узнать есть прекрасный доклад Гриши Кошелева переходите смотрите вот ну вроде как Мы научились данные даже правильно выбирать и отправлять Как нам масштабироваться тут возможны нюансы если нам хватает пропускной способности одного единственного Джаба это прекрасно да нам просто нужно не

Speaker A

допускать повторных запусков наших джив на наших сервисах как это сделать в Джаве есть библиотека Лок например хороший выбор бу изменять данные и проверять после старта дба что наш дж такой уже какой-то экземпляр работает значит нам вот текущий экземпляр не нужен можно его

Speaker A

отложить есть альтернатива для сприн чтобы не подбирать версии работает примерно точно также Вот Но если нам не хватит пропускной способного ис но нам ещ нужно понимать что когда мы работаем с какой нам важен ещ и порядок просто так масштабироваться то есть

Speaker A

выбирать несколько событий разными джабами мы не можем мы можем сначала отправить событие удаления записи а затем событие её обновления и тогда наши потребители Это неправильно обработают Нам такого допускать нельзя Ну и нужно избегать дублирования естественно как такое уметь правильно отправлять события в

Speaker A

нужные партиции чтобы нам не допускать вот это вот разницы если мы научимся на продюсере это делать тогда мы эту проблему решим супер Ну вроде рассказал работает как нужно там настраивать или разрабатывать опять же наверное аудитории вопрос кто не сталкивался с

Speaker A

таким а Неужели нет загоре таком будет делать ВОМ Т нет такого решения Почему Потому что вот то о ЧМ Я не стал подробно рассказывать это механизмы записи в таблицу они очень часто меняются в зависимости от наших кейсов от наших

Speaker A

сервисов от нашей доменной модели Но вот отправку Да с нюансами вот этими которые я проговаривание кака gbc У кого есть инфраструктура разработчики каки очень хорошие люди Они о нас беспокоятся и Они часто используемые кейсы вынесли специальные объекты объекты коннекторов коннекто

Speaker A

бывает два вида коннекторы и Source коннекторы коннекторы вернее Source коннекторы они выбирают данные откуда-то получают данные откуда-то и сохраняют в каку Син коннекторы наоборот Нам нужен в данном случае кавка gbc Source Con он данные выбирает из базы и отправляет в

Speaker A

каку можно развернуть поднять просто как докер образ да Но тогда нужно будет его уметь менедж запускать перезапускать обновлять всё такое настройки вынесены в конфигурацию как и у cdc решений например А да и debezium тоже является условно cdc ой кавка коннектором тоже

Speaker A

а то о чём я говорил чуть раньше jdbc конектор умеет делать по инкремента выбирать данные по тайм смпу выбирать данные может комбинировать можно какой-то кастомный SQL запрос написать можно прямо балк модом выгрузить взять и выгрузить всю таблицу а но опять же у него есть ряд

Speaker A

определённый ряд ограничений например самым главным ну лично для меня для наших команд ограничением стало то что он смещение которое мы хранить можем в таблице он хранит и в топиках Кафки и Ну это не не нерешаемая задача но нужно уметь с ней

Speaker A

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

Speaker A

не быть Да это тоже нужно понимать Наверное я рассказал практически всё что хотел Давайте подводи какие-то краткие итоги А ну во-первых вот такая задача которую я писал она Ну в моей практике Она довольно частая да и у вас я думаю тоже

Speaker A

раз вы сюда пришли что нам нужно отправлять все события с изменением в базе данных а на рынке есть технологии которые проблему решают cdc решение deum кавка Connect если они решают вашу проблему это прекрасно Используйте их не нужно писать велосипеды Да но если они

Speaker A

не подходят по какой-то причине то приходится проблему решать самим А есть простое решение можем просто написать синхрон отправку если у нас это удовлетворяет нашим потребностям и низкие требования к надёжности это подойдёт Если нет то давайте использовать правильные паттерны

Speaker A

transactional Bo хороший способ решить задачу но нужно реализовывать с учётом требований и ограничений наших систем А спасибо за внимание Спасибо Алексей за Такой чудесный доклад который чуть-чуть вымоченное потом и болью Судя по некоторым примерам явно что-то терялось А у нас с вами есть немножко времени на

Speaker A

вопросы также видел задавайте вопросы в чатике а и закидывай я вот вижу Давайте вот прямо первый ряд первый вопрос а чтобы в таблице записи на отправку не застревали в состояни Кет не использовали арек дрек в постгрес не использовали

Speaker A

чтобы сообщения на отправку не застревали очень хороший уровень понимания постгрес не можно использовать можно использовать разные блокировки Да не только пессимистический не только вот такого уровня что мы заблокирую строку и отправим на неё но и то что предлагаешь

Speaker A

Да просто тогда не придётся дополнительных делать статусная модель Да не ну сонно модель она всё но будет но просто она будет подбираться Да просто не придётся сервис делать который будет вычитывать Локи и думать они уже протухли или нет Надо ли их вернуть

Speaker A

назад на отправку да Наверное действительно хороший вопрос так следующий я вот вижу молодой человек уже поднимал руку в красной маечки с серой кофточкой Спасибо большое за доклад вопрос Следующий Можете ли вы короче есть э мнение что вот этот transaction outbox

Speaker A

может быть заменён следующим паттерном паттерн называется Отправь себе сам приходит запрос и ты в сервис который этот запрос обрабатывает отправляет в ту же кафку допустим сообщение ну какое-то что типа пришёл запрос Ну а дальше он он же и

Speaker A

считывает Ну или его сосед Если да И тут такие же гарантии потому что у тебя ретра через ту же самую кафку и так далее и тому подобное ты это можешь всё просить ЕС в следующую каку в которую ты хочешь

Speaker A

положить события э какие-то проблемы Ну и так далее и тому подобное вот просто Я использовал обычно этот паттерн мне интересно ваш опыт э можно Ну то есть использовали ли вы смотрели ли в эту сторону спасибо Ну то есть ты имеешь в

Speaker A

виду что нам нужно не в базу сохранить сообщение сначала а сразу отправить его в кафку затем Вычитать и дальше уже делать что хотим Да да да Это возможно но это вот было что это sec РН это вот это и есть по сути мы переделываем такую

Speaker A

архитектуру по-другому вот да так можно решать понял спасибо так ещё вопросы вот я вижу ди в синей кофточке Спасибо Да ну на само с подобными проблемами Да с подобными проблемами встречались я думаю что ваши вашем решении есть некий недостаток особенно в

Speaker A

обработке ошибочных сообщений вам недостаточно хранить дишни вам записи ошибочные которые есть вам надо уметь сдвигать вниз по выборке потому что вы можете столкнуться с ситуацией когда у вас есть три записи и вы постоянно отправляете три ошибочные записи не

Speaker A

давая шанса переотправить тем которые находятся внизу то есть надо уметь давать шанс остальным иначе у вас просто происходит starvation на переотправить всё правильно но это нужно уметь правильно обрабатывать коллбэк Ну то есть можно Дописать счётчик какой-то да На строке что столько-то раз мы уже

Speaker A

пытались отправить если мы не смогли отправить столько-то раз Мы это в D скинули микрофон Да мы делаем по-другому мы вводим дополнительное поле из разряда tch всякий раз когда мы берём запись в обработку мы берём время записи и соответственно при

Speaker A

селекте Мы всегда сортируем по времени тача получается так что самые неуспешные они у нас запросом уходят в конец давая шанс предыдущим записям быть обработаны да да ну я как бы делал много разного да это не ВС что бы не ВС Что там у нас

Speaker A

существует в наших системах Но это Мы тоже умеем отрабатывать естественно Ну и кажется здесь ещ надо предусматривать вариант того когда у нас допустим есть строгий порядок отправки сообщений Да это тоже такой накладывает давай один вопрос из онлайна US спрашивает А можно ли

Speaker A

объединить решение по сохранению и cdc на отправку событий из таблицы можно только зачем Ну видимо в том что у тебя есть некая табли СМИ кото ты можешь накладывать раскладывать по партиям Азим уже подтягивать из неё данные То есть у

Speaker A

тебя в любом случае оставляется какая-то момент времени когда ты можешь вообще переотправить весь объём сообщений да можно можно но или C решения они нужны чтобы с базой не работать дополнительно То есть тут как бы простота в том чтобы

Speaker A

ничего дополнительно не делать А если делать Ну тогда зачем дези Ну сзиу тоже есть нюансы В зависимости от Вер по так е руки ребята е один вопро для смотрите работали вы с такой ситуацией у Вас например есть сущность очник 1 2 3 у

Speaker A

вас пришли три апдейта на сущность 1 2 3 и смысла три апдейта посылать нет отфильтровать и послать только самый последний апдейт на суе изменения Да Дада суммирующая изменение Не агрегирующие ну То бишь не нагружая кафку и так далее ну в простом

Speaker A

случае Нет я бы сказал Да ну то есть Так можно делать Я уверен что такие кейсы есть но вот в нашем в моём случае таких не было мы делали для упрощения опять же чтобы не делать эти вот агрегации и

Speaker A

понимать когда мы должны отправлять это вот чем-то уже Похоже нао да Как мы там отправляем пачками буферами Ну можно Конечно когда у нас пропускная ность наверное возрастает очень сильно Да у нас много событий Может флин лучше использовать Ну кстати

Speaker A

да вот Лёша тоже накинул момент если ты используешь дебе Зум и какую-то cdc отправку Отправляй пожалуйста все ивенты но ты можешь построить себе какие-то промежуточные в не проце будет слова их по некому модификатору и конеч получатель у тебя

Speaker A

будет уже получать выставленную значению в принципе это умеет довольно быстро и хорошо но ещё может быть кейс Когда нам важно понимать Кто конкретно эту запись изменял то есть эти вот три айдини они могут быть изменены разными пользователями да для аудита Да например

Speaker A

нужно знать кто конкретно в Коре его поменял е вопрос я вот вижу в белой майке молодого человека при Спасибо большое за доклад у меня вот такой вопрос в примере была пессимистические чтобы не брать те же записи а Возможен ли такой вариант чтобы

Speaker A

допустим у нас есть 10 под и Каждый выбирает по ID мод и номер своей поды условно и тогда эти записи не будут пересекаться просто между собой и не нужно будет блокироваться на них а что сделать если у тебя какой-то из подов

Speaker A

вдруг погаснет если не кели и Да если они у тебя не доскей условно Ну что-то случилось такой на инфраструктуре Ну как бы мы живём в парадигме что наши сервисы они живут то постоянно но не всё их количество постоянно тогда как-то всё равно

Speaker A

придётся балансировать между этими моментами А ну вот не знаю Как вариант можно хранить как-то количество вот желаемых Нот Допустим мы хотим 10 нот но в данный момент активно только восемь и тогда на некоторые ноды будет Ну побольше записи подхвата Ну так вот

Speaker A

вообще наверное алгоритм то несложный можно сделать но опять вопрос Нужно ли это надо придётся або что-то синхронизироваться Да чтобы тупо знали о сво количе ни ба может быть Вот кто какой мод читает то есть придётся заводить какую-то Мета информацию Ну то

Speaker A

есть всё равно как бы тут есть нюансы Просто у нас вот в команде фреймворк сделан и вот как раз таки мы очень часто такую практику применяем Вот спасибо Ну если у вас там всё все нюансы об Да если

Speaker A

я ещё доверял бы нашим дата центрам то наверное Мы тоже так сделали во действительно станет не 15 подов а 10 Да это может быть ещё вопросы нука Так у нас есть ещё вопросы из онлайна от ребят А тут вот есть да небольшой не знаю

Speaker A

вопрос или наброс от Алексея самое интересное не рассказали как сделать многопоточная порядок много Ну через партиции и сказали сказали а ну в смысле как детально это сделать как детально Да ну тут уже Инженерная Задачка Давайте писать код отличная Задачка Да кому-то

Speaker A

вю на годовой этот Возможно на все эти но опять же если всё рассказывать К сожалению тема Ну можно углубляться в любом случае углубляться и дальше и дальше Просто тогда уже тайминг доклада вылезет за все разумные пределы ещё один Так вопросы в офлайне

Speaker A

появились я вот вижу две руки есть ещё один вопрос из онлайна но он пря требует наверно более детального ответа и Наверно его в дискуссионная по выбору интервала полинга для опроса базы тут как говорится depends on тут надо подбирать это дело где-то

Speaker A

достаточно ну секунды где-то нужно ускорять а где-то замедлять наоборот вот могуть свой довольно интересный кейс у нас количество подов вдруг резко выросло по другой причине абсолютно но было условно 20 подов стало 200 и очень много началось вот этой гонки потому что

Speaker A

действительно там везде нужно Везде эти жабы запускаются А это почему-то ну не учли забыли и это больно ударило по вакууму даже в очень маленькой таблице потому что shlock он же тоже вакуум пытается Ну как бы обновляет строки А какие значения обычно

Speaker A

используете Ну обычно 200 миллисекунд может быть можно посмотреть на какое-то Я не знаю среднее значение транзакции Да ну надо смотреть в зависимости от того сколько к нам приходит событий и чтобы мы их успевали и отправлять чтобы они накапливались но при этом не сильно

Speaker A

обращались к базе данных Ну 200-250 наверное для какого-нибудь среднего сервиса возможно хватит Да быстрее вряд ли Но если там действительно много их но надо каждые 50 миллисекунд запускать Окей так ребята У нас остался один время на один вопрос в оффлайне ещё рука

Speaker A

Спасибо за доклад вопрос Следующий А какой вообще вы выбирали либо использовали механизм работы с партиции Окей мы там их берём и пишем в них для того чтобы увеличить скорость да обработки сообщений А что с ними потом делать с

Speaker A

партиции в постгрес Ну да Дада они нужны для того чтобы удалять удобно записи после того как они уже обработаны тут Наверное стоит щ не удалять а просто де тачать патиц и дропать всю Не ну там через дроп делается просто дроп партиции

Speaker A

Ну вот да не удалять записи в самой партиции да да Ну это удаление всей партиции соответственно оно не триггерит вакуум у нас весь объём просто данных он удаляется это как ddl операция она не порождает вакуум А за самими партиции вы

Speaker A

как следите Ну то есть ну вот вариант либо через Times db это сделать либо самим написать какой-то крон джоби либо на базе либо в приложении которые Бут запускаться проверять сколько партиции есть ведь партиции тоже это ну объекты база данных

Speaker A

они тоже подаются запросом То есть можно сетнуть какие парти есть на таблице Да понять какие есть ну по неймингу то есть мы придумываем паттерн название и просто по нему понимаем что э таблица партиции есть сейчас там на 7 дней вперёд нам

Speaker A

хватит Да и уже есть на 7 дней назад этого нам уже с избытком мы три самых старых удалим отличный вопрос давай ещ у нас чуть-чуть м осталось очень интерес во монитори табли просто микрометр на гау размер таблицы и вфа с

Speaker A

алертом есть короче делал я специально вот микрометров ский гаш который там каунтер или не помню метрику которая показывает разницу между тем что записано в Counter и последней записью ВБ запускаем Мони через размер таблицы наверное можно можно через разницу между

Speaker A

поступаете Да когда мы можем сделать метрику того сколько мы в базу записываем и можем сделать метрику того сколько мы отправляем разницу между ними сделать и будет Метрика так тоже можно

Topics:Apache KafkaTransactional OutboxCDCгарантированная доставкараспределённые системымикросервисыPostgres vacuumконсистентность данныхраспределённые транзакциипаттерн Сага

Frequently Asked Questions

Что такое паттерн Transactional Outbox и зачем он нужен?

Паттерн Transactional Outbox используется для надёжной отправки событий из базы данных в Kafka, обеспечивая консистентность между сохранением данных и публикацией сообщений.

Какие уровни гарантии доставки сообщений поддерживает Apache Kafka?

Kafka поддерживает три уровня гарантии доставки: максимальная производительность с возможными потерями, at least once с возможными повторами и exactly once с гарантированной единственностью доставки.

Почему важно понимать процесс vacuum в Postgres при работе с Kafka?

Понимание vacuum помогает поддерживать производительность базы данных, так как Postgres использует MVCC, и без регулярной очистки мёртвых строк таблицы могут расти, что влияет на скорость обработки данных.

Get More with the Söz AI App

Transcribe recordings, audio files, and YouTube videos — with AI summaries, speaker detection, and unlimited transcriptions.

App Store Google Play

Or transcribe another YouTube video here →