Поддержка сохранения MQTT в сетке событий Azure

Функция Retain в Message Queuing Telemetry Transport (MQTT) в Azure Event Grid обеспечивает сохранение последнего известного корректного значения темы, которое доступно для новых подписчиков. Эта возможность позволяет новым клиентам мгновенно получать последнее сообщение при подключении, устраняя необходимость ожидать следующей публикации. Это полезно в таких сценариях, как отчеты о состоянии устройства, сигналы управления или данные конфигурации, где своевременный доступ к последнему сообщению является критически важным.

В этой статье представлен обзор работы MQTT Storage, его последствия выставления счетов, ограничения хранения, методы удаления сообщений и рекомендации по управлению.

Выставление счетов

Сохраненная публикация считается двумя операциями MQTT: один для обработки сообщения и один для хранения.

Ограничения хранилища

До 640 МБ или 10 000 сохраненных сообщений на единицу пропускной способности (TU).
Максимальный размер сохраняемого сообщения составляет 64 КБ.

Для более крупных потребностей обратитесь в службу поддержки Azure.

Удаление сообщений

MQTT 3.1.1: Опубликовать пустую полезную нагрузку в тему. Чтобы задать срок действия, обратитесь в службу поддержки Azure.
MQTT 5.0: установите срок действия или отправьте пустое сообщение, чтобы удалить его.

Сохраненные сообщения — получение и список

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

Retain Get: извлекает необработанное хранимое сообщение для конкретного топика. Метаданные сообщения возвращаются через заголовки HTTP.
Список сохраняемых: перечисляет сохраненные сообщения, соответствующие фильтру тем (поддерживаются подстановочные знаки) или пролистывает результаты с помощью маркера продолжения.

Используйте эти API для наблюдаемости, аудита и рабочих процессов в операционной деятельности (например, поддержка устранения неполадок, заполнения задним числом или проверки состояния устройства в масштабе).

Проверка подлинности и авторизация

API "Retain Get" и "Retain List" требуют:

Проверка подлинности: Authorization: Bearer <token>.
Аудитория маркеров или сфера применения: используйте URI или ресурс идентификатора приложения брокера, предоставленный для этой предварительной версии (замените <YOUR TOKEN HERE> в примерах).
RBAC: вызывающий должен иметь разрешение на вызов операций чтения сохраненных сообщений в неймспейсе.

Чтобы получить маркер носителя, выполните следующую команду Azure CLI:

az account get-access-token --resource=https://eventgrid.azure.net --query accessToken -o tsv

API справки

Сохранить Get

GET /mqtt/retainedMessages/message?topic=<YOUR ENCODED TOPIC HERE>&api-version=2025-11-16-preview HTTP/1.1
Authorization: Bearer YOURENTRATOKEN
Host: <YOUR Event Grid MQTT BROKER URL HERE>

Ответ.

Текст: полезные данные необработанного сообщения MQTT (байты).
Заголовки: включают метаданные сообщения (имена, подлежащие изменению в предварительной версии):
- x-ms-mqtt-topic: тема сохраненного сообщения.
- x-ms-mqtt-qos: уровень качества обслуживания (0 или 1).
- x-ms-mqtt-size: полный размер сообщения MQTT (байт).
- x-ms-mqtt-expiry: метка времени истечения срока действия (ISO 8601), если задано.
- x-ms-mqtt-last-modified: последняя измененная метка времени (ISO 8601).

Сохранить список

GET /mqtt/retainedMessages?topicFilter=<YOUR ENCODED TOPIC FILTER HERE>&continuationToken=<MUTUALLY EXCLUSIVE WITH TOPIC FILTER>&maxResults=<1-100>&api-version=2025-11-16-preview HTTP/1.1
Authorization: Bearer YOURENTRATOKEN
Host: <YOUR Event Grid MQTT BROKER URL HERE>

topicFilter поддерживает подстановочные знаки (например, factory/+/state, sensors/#).
continuationToken является взаимоисключающим с topicFilter. Используйте его для продолжения с предыдущей страницы.
maxResults ограничивает размер страницы (1–100).

Ответ (JSON):

{
    "nextLink": "<NULL OR URL HERE>",
    "retainedMessages": [
        {
            "topic": "<SOME TOPIC>",
            "qos": "<SOME QOS>",
            "size": "<FULL MQTT MESSAGE SIZE>",
            "expiry": "<TIME STAMP>",
            "lastModifiedTime": "<TIME STAMP>"
        }
    ]
}

Примеры

Пример кодирования URL-адресов

Тема: factory/line1/cell17/state
- Закодировано: factory%2Fline1%2Fcell17%2Fstate
Фильтр: factory/line1/+/state
- Закодированное: factory%2Fline1%2F%2B%2Fstate

Сохранение примера Get — curl

BROKER="<YOUR BROKER URL HERE>"  # e.g. contoso.westus.ts.eventgrid.azure.net
ENTRATOKEN="<YOUR TOKEN HERE>"
TOPIC_ENC="factory%2Fline1%2Fcell17%2Fstate"

curl -i \
  -H "Authorization: Bearer $ENTRATOKEN" \
  "https://$BROKER/mqtt/retainedMessages/message?topic=$TOPIC_ENC&api-version=2025-11-16-preview"

Пример ответа:

HTTP/1.1 200 OK
x-ms-mqtt-topic: factory/line1/cell17/state
x-ms-mqtt-qos: 1
x-ms-mqtt-size: 42
x-ms-mqtt-expiry: 2025-11-16T08:13:05Z
x-ms-mqtt-last-modified: 2025-11-16T07:59:41Z
content-type: application/octet-stream

{"state":"READY","tempC":25.1,"ts":"2025-11-16T07:59:40Z"}

Сохранение списка с помощью topicFilter — пример curl

BROKER="<YOUR BROKER URL HERE>"
ENTRATOKEN="<YOUR TOKEN HERE>"
FILTER_ENC="factory%2Fline1%2F%2B%2Fstate"

curl -sS \
  -H "Authorization: Bearer $ENTRATOKEN" \
  "https://$BROKER/mqtt/retainedMessages?topicFilter=$FILTER_ENC&maxResults=50&api-version=2025-11-16-preview"