Как использовать пакет MQTT для реализации публикации и подписки

Протокол MQTT обменивается данными посредством обмена предопределенными управляющими пакетами MQTT. Мы возьмем MQTTX в качестве примера, чтобы показать, как реализовать функцию публикации и подписки через пакеты MQTT.

Соединять

Протокол MQTT основан на протоколе TCP/IP. И MQTT-брокеру, и клиенту нужен TCP/IP-адрес.

Маклер

Если у вас пока нет доступного брокера MQTT, EMQX предоставляет адрес публичного брокера для тестирования: broker.emqx.io:1883 .

Клиент

Конфигурация клиента в инструменте MQTTX на самом деле является конфигурацией пакетов Connect в протокол MQTT. Ниже поясняются соответствующие элементы конфигурации:

ID клиента

Сервер использует ClientId для идентификации клиента. Каждый клиент, который подключается к серверу, имеет уникальный ClientId. И клиент, и сервер должны использовать ClientId для определения статуса, связанного с MQTT-сессия между ними.

ClientId должен существовать, но сервер может разрешить клиенту предоставить ClientId с нулевым байтом. Если это сделано, сервер должен рассматривать это как особый случай и присваивать этому клиенту уникальный идентификатор ClientId. Затем он может нормально обработать этот пакет CONNECT.

Имя пользователя Пароль

MQTT может реализовать связанную аутентификацию и авторизацию, отправив имя пользователя и пароль. Однако, если эта информация не зашифрована, имя пользователя и пароль передаются открытым текстом. EMQX не только поддерживает шифрование SSL/TLS, но также обеспечивает emqx-auth-имя пользователя плагин для шифрования паролей.

Сохранить жизнь

Keep Alive — это временной интервал в секундах. Это относится к максимально допустимому интервалу времени между моментом, когда клиент передает управляющий пакет, и временем, когда отправляется следующее сообщение. Клиент несет ответственность за то, чтобы интервал отправки управляющих пакетов не превышал значение keep-alive. Если никакой другой управляющий пакет не может быть отправлен, клиент должен отправить пакет PINGREQ.

Если значение Keep Alive не равно нулю, и сервер не получает контрольный пакет от клиента в течение 1,5-кратного времени Keep Alive, он должен отключить сетевое соединение клиента и считать сетевое соединение отключенным.

Чистая сессия

Клиент и сервер могут сохранять состояние сеанса для обеспечения надежной передачи сообщений по сетевым соединениям. Этот флаг сообщает серверу, является ли это соединение совершенно новым соединением.

Состояние сеанса клиента включает в себя:

• Сообщения уровня QoS 1 и QoS 2, отправленные на сервер, но еще не подтвержденные
• Сообщения уровня QoS 2, полученные от сервера, но еще не подтвержденные

Состояние сеанса сервера включает в себя:

• Существует ли сеанс, даже если остальная часть состояния сеанса пуста.
• Информация о подписке клиента.
• Сообщения уровня QoS 1 и QoS 2, отправленные клиенту, но еще не подтвержденные
• Сообщения уровня QoS 1 и QoS 2 для передачи клиенту.
• Сообщения уровня QoS 2, полученные от клиента, но еще не подтвержденные
• Необязательное сообщение уровня QoS 0 для отправки клиенту.

Если для флага CleanSession установлено значение 1, клиент и сервер должны отменить все предыдущие сеансы и начать новый сеанс. Сеанс длится только до тех пор, пока есть сетевое соединение.

Если для флага CleanSession установлено значение 0, сервер должен возобновить связь с клиентом на основе состояния текущего сеанса (определяемого ClientId). Если с этим идентификатором клиента не связан сеанс, сервер должен создать новый сеанс. Когда соединение разрывается, клиент и сервер должны сохранить информацию о сеансе.

Connack подтверждает запрос на подключение

Когда клиент отправляет пакет Connect, чтобы запросить соединение с сервером, сервер должен отправить пакет Connack в качестве ответа на пакет Connect от клиента. Если клиент не получает пакет CONNACK от сервера в течение разумного времени, клиент должен закрыть сетевое соединение. Разумное время зависит от типа приложения и коммуникационной инфраструктуры. В MQTTXвы можете установить разумное время ожидания с помощью параметра Время ожидания подключения.

Сообщения Connack содержат два важных признака присутствия сеанса и код возврата соединения.

Сессия присутствует

Флаг присутствия сеанса указывает, является ли текущий сеанс новым сеансом. Если сервер получает соединение с флагом CleanSession, равным 1, флаг SessionPresent в пакете Connack равен 0. Если сервер получает соединение с CleanSession 0, значение флага SessionPresent зависит от того, сохранил ли сервер состояние сеанса клиент, соответствующий ClientId. Если сервер сохранил состояние сеанса, флаг SessionPresent в пакете Connack равен 1. Если у сервера нет сохраненного состояния сеанса, флаг SessionPresent в пакете Connack равен 0.

Подключить код возврата

Код возврата соединения представляет собой ответ сервера на это соединение, а 0 указывает, что соединение было принято сервером. Если сервер получает действительный пакет CONNECT, но не может его обработать по какой-либо причине, сервер должен попытаться отправить пакет CONNACK с ненулевым кодом возврата (один в следующей таблице). Если сервер отправляет пакет CONNACK с ненулевым кодом возврата, он должен закрыть сетевое соединение.

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

Клиент отправляет пакет Subscribe на сервер для создания одной или нескольких подписок. Каждый зарегистрированный клиент интересуется одной или несколькими темами. Чтобы пересылать сообщения приложения темам, которые соответствуют этим подпискам, сервер отправляет клиенту пакет публикации. В пакете Subscribe указывается максимальный уровень QoS для каждой подписки, и на его основе сервер отправляет клиенту сообщение приложения.

Полезная нагрузка пакета подписки должна содержать по крайней мере одну пару комбинаций полей тематического фильтра и уровня QoS. Пакет подписки без полезной нагрузки является нарушением протокола.

Использовать MQTTX подключить Брокер broker.emqx.io:1883и создать подписку с темой testtopic/# и Qos равно 2.

Сервер отправляет пакет Subback клиенту, чтобы подтвердить, что он получил пакет Subscribe и обрабатывает его.

Пакет Subback содержит список кодов причин, который используется для указания максимального уровня QoS или ошибки, возникающей для каждой подписки, запрашиваемой пакетом Subscribe. Каждый код причины соответствует фильтру темы в пакете Subscribe. Последовательность кода причины в пакете Subback должна соответствовать порядку тематических фильтров в пакете Subscribe.

Допустимые значения кода возврата:

• 0x00 — максимальное качество обслуживания 0
• 0x01 — успех — максимальное QoS 1
• 0x02 — успех — максимальное QoS 2
• 0x80 — сбой

Публикация сообщений

Пакет публикации относится к сообщению приложения, передаваемому от клиента к серверу или от сервера к клиенту. После получения пакета публикации сервер пересылает сообщение другим клиентам в соответствии с фильтром темы.

Попробуйте использовать MQTTX опубликовать сообщение с темой testtopic/mytopic и содержание{"msg":"hello world"}. Потому что тема testtopic/# была подписана ранее, сообщение, отправленное Брокером, будет получено немедленно.

Тема

Имя темы используется для определения сеанса, в котором должно быть опубликовано сообщение. Имя темы пакета публикации, отправляемого сервером подписавшемуся клиенту, должно соответствовать фильтру темы подписки.

качество обслуживания

QoS относится к уровню качества обслуживания для распространения сообщений приложения.

Пакет публикации не может установить все биты QoS в 1. Если сервер или клиент получает пакет публикации со всеми битами QoS, установленными в 1, он должен закрыть сетевое соединение.

Принцип работы различных уровней QoS см. Введение в протокол MQTT 5.0 — качество обслуживания QoS.

Удерживать

Если для флага RETAIN пакета публикации, отправленного клиентом на сервер, установлено значение 1, сервер должен сохранить это сообщение приложения и его уровень качества обслуживания (QoS), чтобы его можно было передать будущему подписчику с соответствующими именами тем. . При создании новой подписки для каждого соответствующего имени темы, если есть недавно сохраненное сообщение, оно должно быть отправлено этому подписчику. Если сервер получает сообщение Q с флагом RETAIN, равным 1, он должен отбросить все сообщения, ранее сохраненные для этой темы, и рассматривать это новое сообщение как новое сохраненное сообщение для этой темы.

Пакет публикации с флагом сохранения 1 и полезной нагрузкой из нулевых байтов будет рассматриваться сервером как обычное сообщение и будет отправлен клиенту, соответствующему теме подписки. Кроме того, любые существующие сохраненные сообщения в той же теме должны быть удалены, поэтому любые подписчики, отслеживающие эту тему, не получат сохраненное сообщение.

Когда сервер отправляет пакет публикации клиенту, если сообщение отправляется в результате новой подписки клиента, он должен установить флаг сохранения пакета равным 1. Когда пакет публикации отправляется клиенту, поскольку он соответствует установленной подписке, сервер должен установить флаг сохранения в 0, независимо от значения флага сохранения в полученном им сообщении.

Если флаг сохранения пакета публикации, отправленного клиентом на сервер, равен 0, сервер не может сохранить сообщение и не может удалить или заменить любое существующее сохраненное сообщение.

Полезная нагрузка

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

Первоначально опубликовано на