Сообщения возвращаются в очередь до их окончательной обработки

#1

Добрый день.

Инфраструктура:

  • РИБ, Розница 2.2. Через RMQ обменивается пока один магазин. База на узле - серверная, postgresql.
  • В узле одна очередь на прием, другая на отправку.
  • Тип точки публикации – direct, features – durable.
  • Интервал ожидания – 60с, частота пульса – 30с.

Проблема: после получения сообщения из RMQ оно переходит в Unacked. То есть получено, но не подтверждено. Но еще до окончания обработки сообщение становится непринятым (из колонки "Unacked" переходит в "Ready").

Типовой код процедуры получения сообщений:

image
image.png1069×604 31.8 KB

В цикле получаем сообщения, вычленяем тело сообщения, передаем в обработку. В методе " ОбработатьСообщение " вызывается функционал БСП по обмену с РИБ. Сообщения могут быть достаточно большие, т.к. выгружаются большие документы. Соответственно, метод " ПланыОбмена.ПрочитатьИзменения(ЧтениеСообщения, КоличествоЭлементовВТранзакции) " может выполняться достаточно долго, вплоть до 1 часа и более.

И вот когда метод « ПланыОбмена.ПрочитатьИзменения() » находится в процессе выполнения, сообщение из «Unacked» в какой-то момент (по наблюдением – примерно через минуту-две) переходит в «Ready». При этом фоновое задание получение сообщений выполняется и новых не стартует. И когда все же метод « ПланыОбмена.ПрочитатьИзменения() » отработает и таки загрузит сообщение, то, естественно, произойдет ошибка подтверждения при выполнении метода Клиент.BasicAck(), потому что уже нет сообщений для подтверждения. То есть в базу сообщение загружено, но возвращается обратно в очередь, а значит будет загружать повторно.

Заметил, что «откат» сообщения в очередь до обработки происходит, когда соединение RabbitMQ с узлом отваливается (вкладка Connections в админке). Хотя, по идее, ЧастотаПульса (HeartBeat) должна обеспечивать постоянное соединение, пока сообщение уже получено, но еще не подтверждено, насколько я понимаю. Причем с небольшими сообщениями проблем нет – 2 дня обмен шел идеально.

Итак, главный вопрос в том, почему сообщение, даже если долго обрабатывается после получения, не дожидается окончания обработки приложением?

1 Симпатия
#2

heartbeat работает в том же потоке, а не в параллельном. Поэтому, когда 1с долго обрабатывает сообщение, сервер клиенту посылает 2 хертбита, на которые клиент не отвечает и сервер отстреливает соединение. Вариант либо ставить хертбит 0, либо быстрее обрабатывать сообщение, либо на сервере задирать значения хертбита(по умолчанию 60 сек и выше с клиента не поднять).

#3

Я один вижу какой-то авторский комментарий в процедуре “ОбработатьСообщение” :wink:

Но если по сути - то

  1. нужно сокращать время выполнения ПланыОбмена.ПрочитатьИзменения( - вообще-то не должны обмены с РИБ делаться 1 час.
  2. Hearbeat - это такая функция, которая отвечает за то, при каких параметрах мы считаем соединение умершим.

В вашем случае, собственно - пока вы не сократите время выполнения ПлановОбмена, а точнее не повысите частоту выгрузки изменений скажем до 1 минуты, вы можете отключить Heartbeat совсем как функциональность.