RabbitMQ позволяет устанавливать TTL (время жизни) как для сообщений, так и для очередей. Это контролируется дополнительными аргументами очереди, а лучше всего это делать с помощью политики.
TTL сообщения может быть применен к одной очереди, группе очередей или применен на основе сообщения за сообщением.
Настройки TTL также могут применяться с помощью политик оператора.
TTL сообщений в очередях на каждую очередь
TTL сообщения можно установить для данной очереди, задав аргумент message-ttl с помощью политики или указав этот аргумент во время объявления очереди.
Сообщение, которое находится в очереди дольше, чем заданный TTL, считается мертвым. Обратите внимание, что сообщение, направляемое в несколько очередей, может умереть в разное время или вообще не умереть в каждой очереди, в которой оно находится. Смерть сообщения в одной очереди не влияет на жизнь того же сообщения в других очередях.
Сервер гарантирует, что мертвые сообщения не будут доставлены с помощью basic.deliver (потребителю) или включены в ответ basic.get-ok (для разовых операций выборки). Кроме того, сервер будет стараться удалять сообщения по истечении TTL-срока или вскоре после него.
Значение аргумента TTL или политики должно быть неотрицательным целым числом (0 <= n), описывающим период TTL в миллисекундах. Таким образом, значение 1000 означает, что сообщение, добавленное в очередь, будет жить в ней в течение 1 секунды или до тех пор, пока оно не будет доставлено потребителю. Аргумент может иметь тип AMQP 0-9-1 short-short-int, short-int, long-int или long-long-int.
Определение TTL сообщений для очередей с помощью политики
Чтобы задать TTL с помощью политики, добавьте ключ "message-ttl" в определение политики:
rabbitmqctl
| 1 | rabbitmqctl set_policy TTL ".*" '{"message-ttl":60000}' --apply-to queues |
rabbitmqctl (Windows)
| 1 | rabbitmqctl set_policy TTL ".*" "{""message-ttl"":60000}" --apply-to queues |
Это применит TTL в 60 секунд ко всем очередям.
Определение TTL сообщений для очередей с помощью x-аргументов при объявлении
Этот пример на Java создает очередь, в которой сообщения могут находиться не более 60 секунд:
| 1 2 3 | Map<String, Object> args = new HashMap<String, Object>(); args.put("x-message-ttl", 60000); channel.queueDeclare("myqueue", false, false, false, false, args); |
Тот же пример на C#:
| 1 2 3 | var args = new Dictionary<string, object>(); args.Add("x-message-ttl", 60000); model.QueueDeclare("myqueue", false, false, false, false, args); |
Можно применить политику TTL сообщений к очереди, в которой уже есть сообщения, но это связано с некоторыми оговорками.
Первоначальное время истечения срока действия сообщения сохраняется, если оно повторно поставлено в очередь (например, из-за использования метода AMQP с параметром requeue или из-за закрытия канала).
Установка TTL в 0 приводит к тому, что срок действия сообщений истекает по достижении очереди, если они не могут быть доставлены потребителю немедленно. Таким образом, это обеспечивает альтернативу флагу немедленной публикации, который сервер RabbitMQ не поддерживает. В отличие от этого флага, не выдается никаких basic.returns, и если установлен обмен мертвыми буквами, то сообщения будут передаваться мертвыми буквами.
TTL для каждого сообщения
TTL можно задать для каждого сообщения, установив свойство expiration при публикации сообщения.
Значение поля expiration описывает период TTL в миллисекундах. Применяются те же ограничения, что и для x-message-ttl. Поскольку поле expiration должно быть строкой, брокер будет (только) принимать строковое представление числа.
Если указаны TTL для каждой очереди и для каждого сообщения, будет выбрано меньшее значение из двух.
В данном примере используется Java-клиент RabbitMQ для публикации сообщения, которое может находиться в очереди не более 60 секунд:
| 1 2 3 4 5 | byte[] messageBodyBytes = "Hello, world!".getBytes(); AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder() .expiration("60000") .build(); channel.basicPublish("my-exchange", "routing-key", properties, messageBodyBytes); |
Тот же пример на C#:
| 1 2 3 4 5 6 7 8 9 10 | byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes("Hello, world!"); IBasicProperties props = model.CreateBasicProperties(); props.ContentType = "text/plain"; props.DeliveryMode = 2; props.Expiration = "60000"; model.BasicPublish(exchangeName, routingKey, props, messageBodyBytes); |
Предостережения
Очереди, к которым TTL для каждого сообщения был применен задним числом (когда в них уже были сообщения), будут отбрасывать сообщения при наступлении определенных событий. Только когда просроченные сообщения достигают головы очереди, они действительно будут отброшены (или заблокированы). Потребителям не будут доставляться просроченные сообщения. Следует помнить, что между истечением срока действия сообщения и доставкой потребителю может возникнуть естественное состояние гонки, например, срок действия сообщения может истечь после того, как оно было записано в сокет, но до того, как оно дошло до потребителя.
При установке TTL для каждого сообщения просроченные сообщения могут стоять в очереди за непросроченными, пока последние не будут потреблены или просрочены. Следовательно, ресурсы, использованные такими просроченными сообщениями, не будут освобождены, и они будут учитываться в статистике очереди (например, количество сообщений в очереди).
При ретроактивном применении политики TTL для каждого сообщения рекомендуется иметь потребителей в сети, чтобы убедиться, что сообщения отбрасываются быстрее.
Учитывая такое поведение настроек TTL для каждого сообщения в существующих очередях, когда возникает необходимость удалить сообщения для освобождения ресурсов, вместо этого следует использовать TTL очереди (или очистку очереди, или удаление очереди).
TTL очереди
TTL может быть установлен для очередей, а не только для их содержимого. Очереди истекают через определенный период времени только в том случае, если они не используются (например, не имеют потребителей). Эта функция может использоваться вместе со свойством автоматического удаления очереди.
Время истечения срока действия может быть установлено для данной очереди путем задания аргумента x-expires в queue.declare или путем задания политики expires. Это позволяет контролировать, как долго очередь может быть неиспользуемой, прежде чем она будет автоматически удалена. Неиспользуемая очередь означает, что у очереди нет потребителей, очередь не была недавно повторно объявлена (повторное объявление возобновляет аренду), и basic.get не вызывался в течение как минимум периода истечения. Это может использоваться, например, для очередей ответов в стиле RPC, где может быть создано много очередей, которые никогда не будут исчерпаны.
Сервер гарантирует, что очередь будет удалена, если она не используется по крайней мере в течение срока действия. Не гарантируется, как быстро очередь будет удалена после истечения срока действия. Аренда долговременных очередей перезапускается при перезапуске сервера.
Значение аргумента x-expires или политики expires описывает период истечения срока действия в миллисекундах. Оно должно быть целым положительным числом (в отличие от TTL сообщения оно не может быть равно 0). Таким образом, значение 1000 означает, что очередь, которая не используется в течение 1 секунды, будет удалена.
Определение TTL для очередей с помощью политики
Следующая политика заставляет все очереди истекать через 30 минут после последнего использования:
rabbitmqctl
| 1 | rabbitmqctl set_policy expiry ".*" '{"expires":1800000}' --apply-to queues |
rabbitmqctl (Windows)
| 1 | rabbitmqctl.bat set_policy expiry ".*" "{""expires"":1800000}" --apply-to queues |
Определение TTL для очередей с помощью x-аргументов во время декларации
Этот пример на Java создает очередь, срок действия которой истекает после того, как она не используется в течение 30 минут.
| 1 2 3 | Map<String, Object> args = new HashMap<String, Object>(); args.put("x-expires", 1800000); channel.queueDeclare("myqueue", false, false, false, false, args); |