Удержание сообщений

Поскольку объем памяти, доступный для хранения сообщений, ограничен местом на диске, может потребоваться удаление ненужных сообщений. Для периодической очистки очереди сообщений TQE(MQ) использует модуль expirationd. Соблюдение условий для удаления сообщений контролируется функцией удержания сообщений (data retention).

Функция удержания работает согласно установленной политике очистки ненужных сообщений (policy) . Политика может быть включена (policy: cleanup) или отключена (policy: disabled).

При включенной политике сообщения удаляются в зависимости от параметров:

• time - допустимое время жизни сообщения в очереди (в секундах). Значение 0 или отсутствие параметра отключает проверку, сообщения могут находиться в очереди неограниченно долго.
• length - допустимая длина очереди. Значение 0 или отсутствие параметра отключает проверку — длина очереди не ограничена.
• consumers_required - включает очистку по прочитанным сообщениям. При включенной очистке политика сравнивает указанное значение с количеством персистентных подписчиков:
  • Если количество персистентных подписчиков не меньше значения сonsumers_required, то сообщение будет удалено только после того, как его прочтут все персистентные подписчики.
  • Если количество персистентных подписчиков меньше значения сonsumers_required, то сообщение не будет удалено очисткой по прочитанным сообщениям.
  • При consumers_required: 0 и наличии персистентных подписчиков очистка ожидает прочтения сообщения всеми подписчиками.
  • При consumers_required: 0 и отсутствии персистентных подписчиков все сообщения считаются прочитанными и удаляются сразу.

Политика очистки и смежные параметры настраиваются в разделе retention роли roles.tqe-storage. Раздел retention можно настраивать как на глобальном уровне, так и на уровне отдельной очереди. При расхождении параметров приоритет имеет настройка отдельной очереди. Подробнее о настройках раздела retention смотрите в документации по expirationd.

По умолчанию конфигурация для всех очередей выглядит так:

 
roles_cfg:  roles.tqe-storage:    retention:      time: 3153600000             # Допустимое время жизни сообщений всех очередей в секундах.      length: 0                    # Проверка на максимальную длину очереди не выполняется      policy: cleanup              # Политика очистки включена.      options:                     # Конфигурационные параметры для expirationd.start.        iteration_delay: 0         # Максимальное время в секундах для fiber.sleep между итерациями.

Переопределить конфигурацию для определенной очереди можно, указав раздел retention с отличными значениями параметров в разделе конфигурации конкретной очереди example_queue:

 
roles_cfg:  roles.tqe-storage:    queues:      - name: example_queue        retention:          time: 100                  # Допустимое время жизни сообщений конкретной очереди в секундах.          length: 1000               # Максимальная длина очереди указана.          policy: cleanup            # Политика очистки включена.          consumers_required: 2      # Количество постоянных подписчиков, после прочтения которыми сообщение может быть удалено.          options:            tuples_per_iteration: 512

Следующий пример конфигурации позволяет:

• Настроить глобальное время удержания сообщений во всех очередях в 120 секунд и указать максимальную длину очереди в 1024 сообщения.
• Для очереди input:
  • Переопределить время удержания - 60 секунд.
  • Увеличить максимальную длину очереди до 2048 сообщений.
• Для вызова expirationd.start:
  • Указать параметр full_scan_time.
• Для очереди output:
  • Отключить очистку очереди от устаревших сообщений.

При такой конфигурации раздел роли roles.tqe-storage выглядит так:

 
roles_cfg:  roles.tqe-storage:    retention:      time: 120      length: 1024    queues:      - name: input        retention:          time: 60          length: 2048          options:            full_scan_time: 1024      - name: output        retention:          policy: disabled # Политика очистки отключена.

Получение сообщений в TQE(MQ) возможно только при наличии подписки.

Для получения подписки выполните запрос SubscriptionRequest без указания consume_id.

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

Для получения персистентной подписки выполните запрос SubscriptionRequest с указанием параметра consume_id.

С каждым сообщением персистентным подписчикам приходит значение параметра cursor, соответствующее позиции сообщения в очереди. Значение cursor критически важно при восстановлении состояния подписки после отключения.

Восстановить состояние подписки можно при переподключении. Для этого после отключения выполните запрос SubscriptionRequest с теми же значениями параметров queue, routing_key, sharding_key, consume_id, которые использовались при инициализации подписки. В запросе обязательно укажите параметр cursor, который использовался непосредственно перед отключением.

При совпадении параметров в запросе с данными системы подписка восстанавливается, и сообщения отправляются с последней сохраненной позиции. Если же любой параметр не соотвтетствует данным из хранилища, система возвращает ошибку.

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

После перезапуска сервер теряет состояние в памяти. Необходимо переподключиться и отправить запрос SubscriptionRequest, включающий consume_id и cursor последнего полученного сообщения. Сервер автоматически восстанавливает состояние из персистентного хранилища и возобновляет передачу сообщений.

При перезапуске реплики gRPC-сервер автоматически обнаруживает сбой, выбирает другую доступную реплику и возобновляет обработку запросов с последней известной позиции.

После переключения на новую мастер-реплику gRPC-сервер автоматически устанавливает соединение и проверяет состояние подписок. Состояние персистентных подписок полностью восстанавливается на новой реплике.