MQ - Как гарантировать доставку сообщений в нетранзакционной, легкой среде?
Как гарантировать доставку сообщений в нетранзакционной, легкой среде?
Например:
- Обычная ситуация: запись в базу данных, фиксация, отправка сообщения в ZeroMQ | Redis | OtherMQ, потребитель извлекает сообщение, чтобы продолжить обработку...
- 0,05% ситуация: запись в базу данных, фиксация, приложение умирает!, сообщение не отправлено, потребитель не извлекает сообщение, незавершенная обработка.
Как не потерять сообщение (не отправлять сообщение) в этой ситуации?
Редактировать: сообщение должно быть доставлено ровно один раз.
2 ответа
В этом сценарии у вас есть 2 общих ресурса (база данных и очередь), и вы хотите, чтобы они передавались вместе. Вы хотите, чтобы ваша база данных зафиксировала, если сообщение отправлено в очередь. Вы хотите, чтобы ваша база данных не фиксировалась, если она не была успешно отправлена, и наоборот. Это просто глобальный механизм транзакций, такой как 2PC. Однако реализовать механизм глобальных транзакций не так просто, и это также очень дорого.
Я предлагаю вам реализовать по крайней мере одну стратегию на стороне производителя и идемпотентность на стороне потребителя, чтобы обеспечить последовательность.
Вы должны создать таблицу сообщений в базе данных производителя и сохранить сообщения в этой таблице перед отправкой в очередь. Затем с запланированным потоком (здесь может быть несколько потоков для увеличения пропускной способности, но будьте осторожны, если ваши сообщения нужно использовать в том порядке, в котором они были созданы), или с любой другой вещью, которую вы можете отправить их в очередь и пометить как отправленные, чтобы гарантировать, что сообщения, которые уже отправлены, больше не будут отправлены. Даже если вы это сделаете, в некоторых случаях ваши сообщения отправляются более одного раза (например, вы отправляете сообщение в очередь, и ваше приложение не работает, прежде чем пометить сообщение как отправленное). Но это не проблема, потому что мы уже хотим реализовать хотя бы один раз стратегию на стороне производителя, что означает, что мы хотим, чтобы сообщение было отправлено в очередь хотя бы один раз.
Чтобы потребитель не мог потреблять одни и те же сообщения, которые производятся более одного раза на стороне производителя, вы должны реализовать идемпотентных потребителей. Проще говоря, вы можете сохранить идентификатор использованных сообщений в таблице базы данных на стороне потребителя, и перед обработкой сообщений, поступающих из очереди, вы можете проверить, не было ли оно уже использовано. Если он уже используется, вы должны проигнорировать его и получить следующее сообщение.
Конечно, существуют и другие варианты обеспечения согласованности в среде микросервисов. Вы можете найти другие решения в этом замечательном блоге - https://www.nginx.com/blog/event-driven-data-management-microservices/. Решение, которое я объяснил выше, также существует в этом блоге. Вы можете найти его в разделе "Публикация событий с использованием локальных транзакций".
Здесь может быть простой подход.
Предположим, у вас есть транзакция:
- записать данные в БД
- отправить сообщение через ZMQ
- напишите в БД, чтобы отправка была ОК
Поэтому предположим, что ваше приложение завершилось сбоем, пока вы находитесь на шаге 2 или 3. Если это так, вы не знаете, получило ли последнее сообщение очередь клиентов, и вам придется повторно отправлять после перезапуска все сообщения без последнего подтверждения (шаг 3).
Проблема на стороне потребителей, потому что возможно, что они получают сообщение дважды. Чтобы решить эту проблему, вы можете отправить с каждым сообщением идентификатор транзакции, который постоянно увеличивается. Потребитель должен заметить идентификатор транзакции последнего сообщения. Когда входящее сообщение имеет идентификатор транзакции, который не превышает идентификатор транзакции последнего сообщения, сообщение можно игнорировать.
Вопрос в том, можете ли вы изменить структуру сообщения и какой идентификатор транзакции вы можете использовать.