С проблемой консистентности данных мы столкнулись при разработке микросервиса под названием Profile. Он отвечает за регистрацию новых пользователей, хранение данных о них и синхронизируется с монолитной базой. Именно в синхронизации двух баз оказалось несколько проблем.
Синхронизация данных
Чтобы система работала стабильно, когда данные меняются в одной базе данных, изменения должны автоматически отразиться на другой. Для этого мы создали очередь из таких изменений в Kafka.
Мы начали работу с таблицы студентов. К базовым колонкам с именами, фамилиями и классами добавили дополнительные — revision и foreign_revision — для работы с очередью. Теперь при добавлении или изменении данных вызывается триггер в постгресе, который записывает в поле revision текущее время с точностью до миллисекунды. Привожу код добавления колонок и триггера:
ALTER TABLE "students" ADD "revision" timestamp(6) ALTER TABLE "students" ALTER COLUMN "revision" SET DEFAULT timezone('utc', now()); ALTER TABLE "students" ADD "foreign_revision" timestamp(6) CREATE OR REPLACE FUNCTION increase_revision() RETURNS trigger AS $$ BEGIN NEW.revision := timezone('utc', now()); RETURN NEW; END $$ LANGUAGE PLPGSQL; CREATE TRIGGER update_revision BEFORE UPDATE ON students FOR EACH ROW WHEN (old.foreign_revision is not distinct FROM new.foreign_revision and row_to_json(old)::jsonb - 'revision' is distinct FROM row_to_json(new)::jsonb - 'revision') EXECUTE PROCEDURE increase_revision();После добавления студента или изменения его персональных данных формируем и отправляем сообщение в Kafka. Однако если отправить такие сообщения до закрытия транзакции, база пострадает: закончатся коннекты, из-за ошибки сети транзакция откатится. Чтобы этого не происходило, в модели мы использовали after_commit:
after_commit :push_to_exchange, on: [:create, :update]Сервис Profile подписан на общую очередь в Kafka и либо обновляет существующую запись в таблице, либо добавляет новую.
class StudentConsumer def consume(payload, metadata) if record = Student.where(id: payload.id).first record.update!(params(payload)) else Student.create!(params(payload)) end end def params(payload) hash = payload.to_h hash[:foreign_revision] = hash[:revision] hash.slice(*Student.column_names.map(&:to_sym)) end end Таким образом мы добились того, что данные консистентны в двух разных базах. Процесс состоит из четырех шагов:
- Добавляем или обновляем студента в монолите.
- Триггер проставляет текущее время в поле revision.
- Отправляем сообщение в Kafka.
- Получаем сообщение и сохраняем данные о студенте в базе сервиса Profile.
Этот алгоритм работает хорошо, пока не возникнут проблемы: упадет сеть, появятся массовые изменения через update_all или единичные — через update_column, Kafka не будет работать или будет работать медленно.
Чтобы все это не влияло на процесс синхронизации, монолит также подписан на эту очередь и записывает в поле foreign_revision ревизию, которую прочитал из Kafka:
class StudentConsumer def consume(payload, metadata) Student.where(id: payload.id).update_all(foreign_revision: payload.revision) end end Каждые пять минут в монолите запускается воркер, который находит все строки, у которых поля ревизий не совпадают, и заново досылает их в Kafka:
module Profile::SyncShareable def run Student.where("foreign_revision is null or revision != foreign_revision"). where("revision < ?", Time.now - 1.minute). order(revision: :desc). limit(5000). each(&:push_to_exchange) end end Для ускорения этого процесса нужен условный индекс. Он будет маленького размера, потому что у большинства записей ревизии будут совпадать:
CREATE INDEX "index_studends_on_revision" ON "students" ("revision") WHERE revision <> foreign_revisionТаким образом актуальная информация о всех студентах стала доступна для чтения в сервисе Profile. Однако для изменения данных мы были вынуждены ходить в API монолита.
Чтобы вносить изменения прямо в Profile, мы задумались о двусторонней синхронизации.
Двухсторонняя синхронизация
Двустороннюю синхронизацию можно обеспечить, если зеркально повторить весь предыдущий код в сервисе Profile. Но придется решить несколько проблем.
1. Генерация уникального идентификатора
Мы не можем создать нового студента в Profile, если в монолите использован числовой ID. Решит проблему переход на строчный UUID вместо числового инкремента.
2. Синхронизация занимает существенное время
Проблема заключается в том, что данные могут обновляться в двух местах сразу. Например, если в 48 секунд произошло изменение имени в монолите, а в 49 секунд — фамилии в Profile. Теоретически это возможно при исправлениях, дополнениях, автоматической коррекции. Обмен сообщениями через Kafka может занимать дольше трех секунд, и в таком случае изменение имени потеряется из-за более новой версии данных с обновленной фамилией.
Чтобы при двусторонней синхронизации такого не происходило, можно заменить Kafka на что-то более быстрое, например, на RabbitMQ. Но в Kafka хранится журнал транзакций, и мы можем вернуться, проанализировать нашу синхронизацию, в случае аварии проработать транзакции заново. К тому же он умеет работать с двумя разными ЦОД. Для нас это было важно, и мы остались с Kafka. Хотя для кого-то, возможно, актуальнее будет быстрый Rabbit, в котором синхронизация происходит за миллисекунды, а количество воркеров можно регулировать динамически.
3. Асинхронная синхронизация
Когда мы пишем изменения в Profile, нет гарантии, что прочитаем их в монолите, — данные синхронизируются с задержкой. Это надо учитывать, когда разные части приложения написаны поверх разных сервисов. В таких местах приходится отказываться от двусторонней синхронизации и переходить на синхронное взаимодействие через REST API или gRPC.
Таким образом мы можем распределить нагрузку между монолитом и отдельным сервисом, улучшить кодовую базу, делать деплой этого сервиса независимо от монолита.
***
Как вы решали проблему консистентности данных в микросервисной архитектуре? Какой опыт бесшовного распиливания монолита у вас был?
ссылка на оригинал статьи https://habr.com/ru/company/uchi_ru/blog/544854/
Добавить комментарий