Очереди - прекрасный инструмент, который практически идеально масштабируется. Не справляется железо? Просто добавили узлов в кластер. Когда очередь присутствует в проекте, то возникает соблазн всё больше функционала реализовывать с её помощью.
О подводных камнях такого пути поговорим в этой статье.
Рано или поздно, применяя очереди, пользователь сталкивается с вопросом использования их совместно с каким-то сервисом, базой данных и т.п.
-
Заказ выполнен, нужно отправить СМС-уведомление пользователю.
-
Поступил новый заказ, нужно отправить Push-уведомление исполнителям.
-
Работа выполнена, нужно списать деньги со счёта клиента.
Во всех перечисленных примерах изменения в какой-то бизнес-сущности фиксируются в БД (или сервисе с БД), и возникает большой соблазн соблазн отправлять уведомления с использованием очередей.
Что мы имеем в данной ситуации? Первоначальная, простейшая структура кода:
-
Сервис (наша программа) фиксирует изменение данных в БД.
-
Затем сервис кладёт задание в очередь.
Фактически в данном случае требуется реализовать событийный триггер на изменение записи данных.
И в общем случае, получается, здесь мы имеем две записи в две различные БД: сервиса и очереди.
Теперь перенесёмся в реальный мир и рассмотрим, какие ситуации могут возникнуть:
-
Всё в порядке. БД доступна, БД очереди доступна;
-
БД недоступна, БД очереди доступна;
-
БД доступна, БД очереди недоступна;
-
БД недоступна, БД очереди недоступна.
Наиболее опасен для бизнеспроцессов тут третий случай: после того, как данные были зафиксированы в БД, уведомить заинтересованные стороны не получилось.
И вот, если начинать задумываться над этой проблемой, то реализация становится из очень простой... очень сложной. В пределе приходится реализовывать двухфазный коммит.
Но со сложными решениями всё понятно, давайте подумаем о более простых.
Неправильные решения
Очень многие, впервые столкнувшись с этой проблемой (как правило, происходит это, когда система реализована и уже значительное время работает), начинают с того, что предлагают следующее решение:
-
Открываем транзакцию в БД.
-
Изменяем запись в БД.
-
Записываем в очередь.
-
Закрываем транзакцию в БД.
В этом случае, если одна из БД недоступна, то и в другую запись не запишется.
Однако тут имеются следующие проблемы:
-
Данное решение условно работоспособно только в том случае, если обработчик очереди никогда не будет обращаться к записи в БД. Поскольку обработчик очереди может взять задачу на исполнение между пунктами 3 и 4 (Записи в БД ещё не зафиксированы).
-
Закрытие транзакции также может завершиться с ошибкой. В этом случае задание в очереди останется неотменённым.
Выполнение всей работы в обработчике
Простейшее решение.
Вместо изменения записи кладём в очередь задание о том, что надо это сделать. Обработчик очереди выполняет оба действия: и изменения записи в БД, и триггерное. Это решение сложнее, чем первоначальное, но всё равно довольно простое.
Проблема в том, что инициирующему действие коду может быть нужен результат работы (например, для того, чтобы вернуть http-код успеха/ошибки).
Дополнительная таблица в БД
Более простое решение, нежели двухфазный коммит.
В игре у нас появляется дополнительный процесс/демон,
перекладывающий задачи из дополнительной таблицы БД (назовём её
queue) в очередь.
В этом случае запросы, модифицирующие БД, пишут записи в две таблицы в той же транзакции (или в том же запросе):
/* было */UPDATE "orders"SET "status" = 'complete'WHERE "order_id" = $1RETURNING *
/* стало */WITH "o" AS ( UPDATE "orders" SET "status" = 'complete' WHERE "order_id" = $1 RETURNING *),"q" AS ( INSERT INTO "queue" ( "key", "data" ) SELECT "o.order_id", "o.status" FROM "o")SELECT *FROM "o"
Примечание: Код записи в таблицу
queue можно вынести в триггер уровня БД, а также
спроектировать так, чтобы сообщения были универсальными и подходили
не только для orders.
Если код, перекладывающий записи из таблицы queue в
очередь, не использует двухфазный коммит, то к обработчику задания
в очереди появляется требование идемпотентности.
Это происходит потому, что при сбоях возможно появление дублей
заданий в очереди.
Простейший алгоритм цикла демона:
-
Взять задачу из таблицы
queue. -
Положить её в очередь.
-
Удалить задачу в таблице.
Локальный лог/кеш сообщений
Если потеря данных для нас только неприятна, но не критична, то в целом можно ограничиться первоначальным решением. Однако, если частота возникающих проблем всё-таки заставляет обратить на себя внимание (пользователи жалуются на периодически зависающий интерфейс, на неполучение СМС и т.п.), то можно применить следующий подход:
-
резко понизить вероятность неуспеха при отправке сообщения в очередь.
Для этого, как в предыдущем варианте, применяем промежуточное
кеширование записей сообщений для очереди, но пишем их не в
дополнительную таблицу, а в локальный O_APPEND-файл
или локально запущенную очередь.
-
сервис (наша программа) фиксирует изменение данных в БД,
-
затем сервис кладёт задание в локальный лог.
-
Дополнительный демон перекладывает сообщения из этого лога (или локальной очереди) в общую очередь.
Запись в локальный файл коротких сообщений завершается ошибкой в крайне редких ситуациях, и поэтому можно считать, что проблема, возникающая крайне редко, есть решённая проблема.
Итоги
Как видим, вариантов решения проблемы немного. Если мы хотим сохранять систему простой (KISS-принцип), то введение дополнительного демона и кеша/лога сообщений в БД или локальном файле/БД даст небольшое увеличение сложности. При этом очень важно сохранять обработчик идемпотентным, поскольку при сбоях в момент перекладывания заданий из локального кеша в общую очередь возможно появление дублей.
А обобщённое решение предполагает использование двухфазного коммита.
