MQTT протокол#

Описание#

MQTT — это легковесный клиент-серверный протокол, предназначенный для обмена сообщениями по модели публикация-подписка. MQTT был специально разработан для сокращения транспортных накладных расходов (и, следовательно, сетевого трафика) и запрогораммированного футпринта на клиентских устройствах. По этой причине MQTT идеально подходит для ограниченных устройств, таких как датчики и приводы, и он быстро стал стандартным протоколом коммуникации defacto для IoT.

Apache Activemq Artemis поддерживает следующие версии MQTT (со ссылками на их соответствующие спецификации):

• 3.1

• 3.1.1

• 5.0

По умолчанию в ActiveMQ Artemis есть акцепторы, настроенные для принятия соединений MQTT на портах 61616 и 1883.

Полное использование данного протокола приведено в опенсорсной документации Apache Activemq Artemis.

Общие сведения об MQTT#

Что нужно знать о MQTT?#

1. MQTT не использует адреса и очереди, mqtt использует топики.

• топик (Topic) - эндпоинт на брокере, в который пишут продюсеры и из которого читают консюмеры;

• топики можно огранизовывать в иерархии типа sberbank/Controller/<serial_number>/In/Value с помощью /, это пригодится в wildcard подписках;

• клиенты (продюсеры и консюмеры) отправляют сообщения в топики. Топики именуются с помощью знака «/», например mqtt/test/topic. Брокеры SMB(Apache ActiveMQ Artemis) работают в парадигме адресов и очередей. При этом имя топика, в который пишет продюсер преобразуется в multicast адрес, где символ «/» заменяется на «.», например mqtt.test.topic.

2. MQTT-консюмеры не читают очереди, они подписываются на топики (Subscription).

• консюмер при подписке к топику («имя топика») создает в адресе («имя адреса») очередь с названием «<clientID для консюмера>.<имя адреса>», например consumer1.mqtt.test.topic

3. MQTT использует концепцию publish/subscribe:

• топики/подписки не создаются до подключения клиентов;

• несколько консюмеров могут подписаться на один и тот же топик и получать одни и те же сообщения;

• консюмеры получают только те сообщения, которые были отправлены в топик после их подписки. Если в топике нет консюмеров - сообщения теряются.

4. Настройка client.id в MQTT важна:

• на ее основе создаются очереди для клиентов;

• несколько клиентов с одинаковым client.id не могут работать одновременно, новый клиент отключает старый.

5. MQTT имеет несколько разных версий, используются две версии: 3.1.1 и 5.

Дополнительные возможности для MQTT 5:

• появилась простая форма управления потоком (flow control);

• введено использование механизма числовых псевдонимов топиков (topic aliasing);

• поддержка shared-подписок;

• введен параметр «максимальный размер пакета» (maximum packet size);

• MQTT 5 и MQTT 3 используется различная конфигурация очистки сессий (CleanSession) Добавлена возможность использовать заголовки для MQTT-сообщений (MQTT 3 - заголовков у сообщений нет)

6. MQTT поддерживает только байтовые сообщения Настройки форматов существуют, но Apache ActiveMQ Artemis их не использует:

• payloadFormatIndicator (UTF-8, UNCPECIFIED) - главный кандидат на байтовое/текстовое сообщение, по спеке должен валидировать формат на брокере;

• content-type - заголовок с mime-типом, по спеке игнорируется брокером.

7. MQTT не поддерживает large messages, они конвертируются в regular.

8. MQTT может создавать дополнительные системные очереди:

• $sys.mqtt.sessions - хранит данные persistent-сессий

• $sys.mqtt.retain.<имя_топика> - retain-сообщения для топиков

• $sys.mqtt.queue.qos2.<имя_топика> - управляющие очереди для exactly once (QoS 2) семантики доставки сообщений

Виды подписок#

1. Обычная подписка

Основной сценарий, в котором продюсер отправляет сообщения в топик mqtt/test/topic, один или несколько консюмеров подписываются на этот же топик. Для отправки сообщений на другой брокер должен использоваться механизм балансировки через адрес.

2. WildCard подписка

Вид подписки, при котором консюмер может подписаться сразу на несколько топиков с помощью wildcard синтаксиса, например:

• mqtt/test/+;

• mqtt/+/topic;

• mqtt/#.

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

В UI символ «*» будет отображаться с символом «" перед ним в качестве экрана.

3. Shared подписка

Одна подписка на несколько топиков делится между несколькими клиентами в группе, при этом сообщения из топика распределяются между клиентами алгоритмом round-robin(или иным внутренним алгоритмом Apache ActiveMQ Artemis). Сообщения должны распределяться между консюмерами в одной группе и дублироваться для разных групп. Для отправки сообщений на другой брокер должен использоваться механизм балансировки через адрес. Актуально для версии MQTT 3.1.1 в Apache ActiveMQ Artemis.

4. Shared-WildCard подписка

Одна подписка на несколько топиков делится между несколькими клиентами в группе, при этом сообщения из топика распределяются между клиентами алгоритмом round-robin.

Дополнительное пояснение к схеме: Создается новый адрес mqtt..topic на каждом брокере в кластере, который соответствует mqtt/+/topic. Создается новая очередь group1.mqtt..topic в новом адресе на каждом брокере. При этом всем адресам (mqtt.test.topic, mqtt.other.topic, mqtt..topic) присваивается Queue Name = group1.mqtt..topic, а также RemoteQueueNames = group1.mqtt.*.topic, …

В UI символ «*» будет отображаться с символом «" перед ним в качестве экрана.

5. Одновременная обычная и shared подписки

Консюмер и группа консюмеров одновременно подписываются на один топик. Группы консюмеров и отдельные консюмеры должны получать дубликаты сообщений.

Дополнительные возможности MQTT#

1. QOS

QOS (Quality of service) настраивается на клиенте и отвечает за семантику доставки:

• 0 AT_MOST_ONCE - сообщение возможно будет доставлено один раз, без подтверждений об отправке/получении. При любых проблемах доставки сообщение теряется;

• 1 AT_LEAST_ONCE - сообщение будет гарантированно доставлено один раз или более, используются подтверджения об отправке/получении. При проблемах доставки сообщение будет переотправлено, что может породить дубли сообщения;

• 2 EXACTLY_ONCE сообщений будет гарантированно доставлено ровно один раз, без дублей. Для этого режима дополнительно используются системные очереди $sys.mqtt.queue.qos2.<имя_топика>.

2. Flow Control

Ограничение кол-ва сообщений при одновременной асинхронной отправке, только для MQTT 5.

• брокер может сообщить клиенту, сколько сообщений QoS 1 и QoS 2 он может получить до подтверждения, и наоборот;

• эта настройка регулируется на брокере путем установки URL-параметра receiveMaximum для MQTT-акцептора в broker.xml;

• значение по умолчанию равно 65 535 (максимальное значение 2-байтового целого числа, используемого MQTT);

• значение, равное 0, запрещено спецификацией MQTT 5;

• значение -1 не позволит брокеру информировать клиента о максимальном количестве получаемых данных, что означает, что управление потоком данных от клиентов к брокеру будет отключено. По сути, это то же самое, что установить значение 65 535, но уменьшает размер пакета CONNACK на несколько байт.

3. Retain messages

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

• для хранения используются служебные retain-очереди $sys.mqtt.retain.<имя_топика>;

• retain-сообщения хранятся пока клиент явно их не удалит или не истечёт (expires) время их жизни;

• за автоудаление пустых служебных retain-очередей отвечает настройка из address-settings:

    <address-setting match="$sys.mqtt.retain.#">
        <auto-delete-queues>true</auto-delete-queues>
        <auto-delete-addresses>true</auto-delete-addresses>
    </address-setting>

4. Will messages

Отдельный вид сообщений, который может отправляться на брокер в момент подключения клиента.

• вместе с сообщением указываются топики, в которые отправляется сообщение;

• сообщение отправляется на указанные топики в момент отключения консюмера, который отправил will-сообщение.

Параметры брокера для работы с MQTT#

Тонкости работы с MQTT#

Для добавления роли пригодной для MQTT в конфигурационном файле vars.yml в блоке artemis_roles нужно задать параметры:

- user: CN=myname, OU=00CA, O=SBRF, L=Moscow, ST=Moscow, C=RU 
  topic: this/is/#
  type: consume,browse,send

, где

• user - DN пользователя;

• topic - имя топика;

• type - тип операции.

В зависимости от комбинации типов в security-settings.xml будут добавлены разрешения:

  <security-setting match="this.is.#">
    <permission type="createNonDurableQueue" roles="consume_this_is_#"/> 
    <permission type="deleteNonDurableQueue" roles="consume_this_is_#"/> 
    <permission type="createDurableQueue" roles="consume_this_is_#"/> 
    <permission type="deleteDurableQueue" roles="consume_this_is_#"/> 
    <permission type="createAddress" roles="send_this_is_#,consume_this_is_#"/> 
    <permission type="deleteAddress" roles="consume_this_is_#"/>
    <permission type="consume" roles="consume_this_is_#"/> 
    <permission type="browse" roles="browse_this_is_#"/> 
    <permission type="send" roles="send_this_is_#"/>
  </security-setting>

Для корректной работы MQTT протокола для адресов в конфигурационном файле vars.yml в блоке artemis_addresses необходимо указать настройки:

artemis_addresses:
  defaults: # дефолтные значения конфигураций для адресов
    default_address_routing_type: 'MULTICAST' # routing_type:MULTICAST - тип маршрутизации, используемый для автоматически созданных адресов
    auto_create_queues: 'true' # bool:true - создавать ли автоматически очередь, когда клиент отправляет сообщение или пытается потребить сообщение из очереди
    auto_create_addresses: 'true' # bool:true - создавать ли автоматически адреса, когда клиент отправляет сообщение или пытается потребить сообщение из очереди, сопоставленной с адресом, который не существует
    auto_delete_queues: 'true' # bool:true - удалять ли автоматически созданные очереди, когда очередь имеет 0 потребителей и 0 сообщений
    auto_delete_addresses: 'true' # bool:true - удалять ли автоматически созданные адреса, когда они больше не имеют очередей