Шаблон исходящих сообщений - как мы можем предотвратить процесс дублирования сообщений в процессе ретрансляции сообщений?

Question

Обычный способ реализации шаблона исходящих сообщений состоит в том, чтобы сохранить полезную нагрузку сообщения в таблице исходящих сообщений и иметь отдельный процесс ( Message Relay ) для ожидающих сообщений и публиковать их. в брокер сообщений, Кафка в моем случае.

Состояние исходящей таблицы может быть таким, как показано ниже.

 OUTBOX TABLE
 ---------------------------------
|ID | STATE     | TOPIC | PAYLOAD |
 ---------------------------------
| 1 | PROCESSED | user            |
| 2 | PENDING   | user            |
| 3 | PENDING   | billing         |
----------------------------------

My Message Relay - это приложение Spring Boot / Cloud Stream, которое периодически (@Scheduled) ищет записи PENDING, публикует их в Kafka и обновляет запись в состоянии PROCESSED.

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

Другая ситуация : предполагается, что только один ретранслятор сообщений. Он получает одну запись в ожидании, публикует ее в теме, но вылетает, прежде чем обновить запись до ОБРАБОТАНО. Когда он снова запустится, он найдет ту же запись PENDING и снова опубликует ее. Есть ли способ избежать этого дублирования или единственный способ - разработать идемпотентную систему.

Answer 1

Чтобы предотвратить первую проблему, вы должны использовать блокировку базы данных.

SELECT * FROM outbox WHERE id = 1 FOR UPDATE

Это предотвратит доступ других процессов к той же строке.

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

Таким образом, одним из способов может быть установление записи в состояние, такое как ОБРАБОТКА, перед отправкой его в Kafka, и если приложение падает, вы должны проверить, есть ли записи в состоянии ОБРАБОТКА, и выполнить некоторую задачу очистки, чтобы выяснить, были ли они уже отправлено в Кафку.

Но лучшим решением было бы иметь идемпотентную систему, способную обрабатывать дубликаты.