Проект MQTTX

Китайский | Английский

• 1 Введение
  • 1.1 Быстрый старт
  • 1.2 Зависимости проекта
  • 1.3 Онлайн примеры
• 2 Архитектура
  • 2.1 Структура каталогов
• 3 Запуск с помощью Docker
• 4 Описание функциональности
  • 4.1 Поддержка QoS
  • 4.2 Поддержка topic filter
  • 4.3 Поддержка кластера
  • 4.4 Поддержка SSL
  • 4.5 Поддержка безопасности тем
  • 4.6 Поддержка общих тем
  • 4.7 Поддержка WebSocket
  • 4.8 Системные темы
    • 4.8.1 Темы состояния
    • 4.8.2 Темы функций
  • 4.9 Поддержка моста сообщений
  • 4.10 Поддержка ограничения скорости тем
  • 4.11 Поддержка сохранения сообщений
  • 4.12 Поддержка базовой аутентификации
• 5 Сообщение от разработчиков
• 6 Приложения
  • 6.1 Настройки конфигурации## 1 Введение

Mqttx разработан на основе протокола MQTT v3.1.1 с целью предоставления лёгко используемого и высокоэффективного broker'а MQTT.

Примечание: ветка v1.2 требует JDK17, остальные ветки требуют JDK8

Связанные проекты: Mqttx-Client реализация и использование очень простого клиента MQTTv3.1.1.

1.1 Быстрый старт

Хотите быстро попробовать через Docker? Смотрите Запуск с помощью Docker

1. Упаковка

   • Разработка:
     1. Запустите redis экземпляр
     2. Выполните команду mvnw -P dev -DskipTests=true clean package

2. Запуск

   1. Запустите команду: java -jar mqttx-1.0.5.BETA.jar

Иллюстрация:

1.2 Зависимости проекта

• Redis: кластерные сообщения, сохранение сообщений
• Kafka: поддержка моста сообщений, кластерные сообщения (необязательная функция)

Другие примечания:1. Проект использует lombok, для использования с IDE установите соответствующие плагины.

Рекомендованное средство разработки — IntelliJ IDEA

Пример: в idea требуется установка плагина Lombok, а также активация Enable annotation processing через settings > Build, Execution, Deployment > Compiler > Annotation Processor.

1.3 Онлайн примеры

Облачные сервисы истекли, доступ к экземплярам невозможен, есть ли друзья, готовые помочь? /(ㄒoㄒ)/~~> В облаке запущено одноэкземпельное служение mqttx, доступное для функционального тестирования:

1. Поддерживает работу без SSL.
2. Активирован WebSocket, можно использовать для тестирования через http://ws.tool.tusk.link/, заменив домен на 119.45.158.51 (порт и адрес остаются без изменений).
3. Поддерживает функцию совместного подписывания.
4. Версия развертывания v1.0.6.RELEASE.

2 Архитектура

mqttx поддерживает аутентификацию клиентов и управление правами доступа к темам публикаций/подписок, если требуется использование вместе, рекомендуется следующая архитектура:

Сервис аутентификации клиентов реализуется пользователем самостоятельно

Внутренняя реализация (перечислены ключевые компоненты):

2.1 Структура каталогов```

├─java │ └─com │ └─jun │ └─mqttx │ ├─broker # Реализация протокола MQTT и обработка сообщений │ │ ├─codec # Кодирование и декодирование │ │ └─handler # Обработчики сообщений (публикация, подписка, соединение и т.д.) │ ├─config # Конфигурация, главным образом объявление бинов │ ├─constants # Константы │ ├─consumer # Потребители сообщений в кластере │ ├─entity # Энтитеты │ ├─exception # Исключения │ ├─service # Интерфейсы бизнес-сервисов (аутентификация пользователей, хранение сообщений и т.д.) │ │ └─impl # По умолчанию реализация │ └─utils # Утилиты └─resources # Ресурсы (файл application.yml находится здесь) ├─META-INF # Дополнительные конфигурации Spring └─tls # Адрес расположения CA

Изображение загружено на **Docker-Hub**, доступ к нему: [fantasywujun/mqttx - Docker Hub](https://hub.docker.com/r/fantasywujun/mqttx) все изображения. После установки окружения Docker выполните команду `docker-compose -f ./docker-compose.yml up` для запуска:

Результат запуска показан ниже:

![y3R3tI.md.png](https://s3.ax1x.com/2021/02/04/y3R3tI.md.png)

| Команда docker pull                      | Описание                                                                                           |
| ----------------------------------------- | -------------------------------------------------------------------------------------------------- |
| `docker pull fantasywujun/mqttx:1.2.0`   | Версия `mqttx:1.2.0` на основе `jdk17.0.1`                                                         |
| `docker pull fantasywujun/mqttx:1.2.1`   | Версия `mqttx:1.2.1` на основе `jdk17.0.1`                                                         |
| `docker pull fantasywujun/mqttx:1.2.2`   | Версия `mqttx:1.2.2` на основе `jdk17.0.1`                                                         |
| `docker pull fantasywujun/mqttx:1.2.3`   | Версия `mqttx:1.2.3` на основе `jdk17.0.1`                                                         |

**docker-compose** файл содержит следующее:

```yaml
version: "2"
services:
  redis:
    container_name: redis-for-mqttx
    image: redis
  mqttx:
    container_name: mqttx
    image: fantasywujun/mqttx:1.2.2
    environment:
      mqttx.max-bytes-in-message: 10485760
      mqttx.web-socket.enable: false
    ports:
      - 1883:1883

4 Функциональные характеристики

4.1 Поддержка QoS

Для поддержки QoS1 и QoS2 используется Redis в качестве хранилища данных. Эта часть уже реализована как интерфейсы, которые можно заменить своими реализациями (например, с использованием MySQL).

4.2 Поддержка topicFilter1. Поддерживает многократные шаблоны # и одиночные шаблоны +.

2. Неподдерживаемый формат темы заканчивается слешем /, например a/b/. Вместо этого используйте a/b.
3. Другие правила см. в разделе MQTT v3.1.1 4.7 Тема и фильтрация темы.> Mqttx проверяет корректность подписки на тему, но не проверяет корректность темы при публикации. Это можно ограничить через 4.5 поддержка безопасности темы.

Примеры:

Инструмент для проверки находится здесь: com.jun.mqttx.utils.TopicUtils.

4.3 Поддержка кластера

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

• Kafka – рекомендованная конфигурация для лучшей производительности. Kafka используется в качестве распространителя сообщений для кластера.
• Redis – по умолчанию.

Реализация принципа представлена ниже:

1. mqttx.cluster.enable – ключевой параметр, значение по умолчанию false.
2. mqttx.cluster.type – тип промежуточного звена, значение по умолчанию redis.

Примечания:

1. В версиях до v1.0.5.RELEASE функционал кластера имеет ошибки и недоступен.
2. Для использования kafka как кластерного распространителя сообщений требуется ручное изменение конфигурации файла application-*.yml. Пример конфигурации можно найти в разделе 3. kafka кластер файла application-dev.yml.#### 4.4 Поддержка SSL

Для активации SSL вам потребуется наличие CA (самозаверенного или приобретённого сертификата). Затем следует изменить несколько параметров в файле application.yml:

1. mqttx.ssl.enable – ключевой параметр, значение по умолчанию false, управляет состоянием SSL для websocket и socket.
2. mqttx.ssl.key-store-location – путь к keystore, относительно classpath.
3. mqttx.ssl.key-store-password – пароль для keystore.
4. mqttx.ssl.key-store-type – тип keystore, например PKCS12.
5. mqttx.ssl.client-auth – требование проверки клиентского сертификата серверной стороной, значение по умолчанию NONE.

Файл mqttx.keystore в директории resources/tls предназначен для тестирования, пароль: 123456.

Инструмент для загрузки сертификатов находится в классе com/jun/mqttx/utils/SslUtils.java.

4.5 Поддержка безопасности тем

Чтобы ограничивать подписку на темы клиентами, внедряется механизм проверки прав доступа к подписке и публикации тем:

1. mqttx.enable-topic-sub-pub-secure – ключевой параметр, значение по умолчанию false.

2. При получении брокером соединения (conn) запроса, он запрашивает {clientId, username, password} у mqttx.auth.url. Ответ от этого интерфейса содержит объект с полями authorizedSub, authorizedPub, содержащими список авторизованных тем для подписки и публикации соответственно.

   Подробнее см. раздел 4.12 Базовая поддержка аутентификации.3. Брокер проверяет права клиента при каждом событии подписки и публикации.

Поддерживаемые типы тем:

• Обычная тема
• Общая тема
• Системная тема

4.6 Поддержка общих тем

Общий подписчик является частью протокола mqtt5. MQTTX реализует его согласно стандарту.

1. Формат: $share/{ShareName}/{фильтр}, где $share — префикс, ShareName — имя общего подписчика, фильтр — фильтр неподключенного подписчика.
2. Поддерживаются следующие правила распределения сообщений:
   1. round – циклическое распределение.
   2. random – случайное распределение.
3. Поддерживается группировка подписчиков по имени общего подписчика.

Для получения более подробной информации обратитесь к протоколу MQTT Version 5.0 (oasis-open.org).

Ниже приведена схема, демонстрирующая различия между общими темами и обычными темами:

Стратегия распространения сообщений msg-a зависит от конфигурационного параметра mqttx.share-topic.share-sub-strategy.

Эта стратегия может использоваться вместе с сессией cleanSession = 1, при которой клиент, использующий общую тему, будет удален из подписок после отключения соединения сервером. Это означает, что сообщения будут доставлены только активным клиентам.Описание CleanSession: согласно протоколу MQTT 3.1.1, если cleanSession = 1, то при отключении соединения клиент и сервер обязаны удалить любую предыдущую сессию и начать новую. Эта сессия продолжается до тех пор, пока существует сетевое соединение. Данные состояния этой сессии не должны повторно использоваться в последующих сессиях [MQTT-3.1.2-6].Состояние сессии клиента состоит из следующего:

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

Состояние сессии сервера состоит из следующего:

• Существование сессии, даже если остальное состояние сессии пусто;
• Подписки клиента;
• Сообщения QoS 1 и QoS 2, отправленные на клиента, но ещё не полностью подтверждены;
• Сообщения QoS 1 и QoS 2, ожидающие передачи на клиента;
• Сообщения QoS 2, полученные от клиента, но ещё не полностью подтверждены;
• В качестве опции, сообщения QoS 0, ожидающие передачи на клиента.

4.7 Поддержка WebSocket

Поддерживается

4.8 Системные темы

mqttx брокер имеет встроенные системные темы, которые могут быть использованы пользователями по мере необходимости.

Системные темы не поддерживают следующие возможности:

• Кластеры: система тем не поддерживает кластеры, включая сообщения и подписки;
• PERSISTENCE: сообщения системных тем не поддерживают PERSISTENCE, включая отношения подписок;
• QoS: не поддерживает QoS 1 и QoS 2, поддерживает только QoS 0.

Внимание: механизмы безопасности тем также влияют на возможность подписки клиентов на системные темы; незарегистрированные клиенты не смогут подписываться на системные темы. Системные темы можно разделить на два типа:1. Темы состояния: отображают состояние broker. 2. Функциональные темы: предоставляют внешнюю функциональную поддержку.

4.8.1 Темы состояния

Клиенты могут получать состояние broker, подписываясь на системные темы. В настоящее время система поддерживает следующие темы состояния:| Тема | Описание | | --------------------------------------- | ---------------------------------------------------------- | | $SYS/broker/{brokerId}/status | Подписчики этой темы будут регулярно (через mqttx.sys-topic.interval) получать состояние broker, которое включает все значения ниже.
Примечание: После отключения соединения клиентская подписка будет автоматически отменена. | | $SYS/broker/activeConnectCount | Немедленно возвращает текущее количество активных подключений.
Триггер: каждое новое подключение триггерит событие. | | $SYS/broker/time | Немедленно возвращает текущий временной штамп.
Триггер: каждый новый запрос триггерит событие. | | $SYS/broker/version | Немедленно возвращает версию broker.
Триггер: каждый новый запрос триггерит событие. | | $SYS/broker/receivedMsg | Немедленно возвращает общее количество принятых сообщений MQTT с момента запуска broker (не считая ping).
Триггер: каждый новый запрос триггерит событие. | | $SYS/broker/sendMsg | Немедленно возвращает общее количество отправленных сообщений MQTT с момента запуска broker (не считая pingAck).
Триггер: каждый новый запрос триггерит событие. | | $SYS/broker/uptime | Немедленно возвращает время работы broker в секундах.
Триггер: каждый новый запрос триггерит событие. | | $SYS/broker/maxActiveConnectCount | Немедленно возвращает максимальное количество активных TCP-соединений с момента запуска broker. |
Триггер: каждый новый запрос triggeрит событие.|В системной теме $SYS/broker/{brokerId}/status значение brokerId является конфигурационным параметром (см. 6.1 Конфигурация) и может использоваться вместе с темой $SYS/broker/+/status.

Формат ответа представлен в виде строки JSON:

{
    "activeConnectCount": 1,
    "maxActiveConnectCount": 2,
    "receivedMsg": 6,
    "sendMsg": 77,
    "timestamp": "2021-03-23T23:05:37.035",
    "uptime": 149,
    "version": "1.0.7.RELEASE"
}

4.8.2 Функциональные темы

Этот функциональный запрос был создан в задаче: Отслеживание состояния MQTT-клиента (в сети, вне сети) · Issue #8 · Amazingwujun/mqttx (github.com)| Тема | Описание | | -------------------------------------------------------------------------------------- | ------------------------------------------------------------ | | $SYS/broker/{brokerId}/clients/{clientId}/connected | Тема уведомления о подключении клиента
Триггер: когда клиент подключается, broker отправляет сообщение на эту тему | | $SYS/broker/{brokerId}/clients/{clientId}/disconnected | Тема уведомления о отключении клиента
Триггер: когда клиент отключается, broker отправляет сообщение на эту тему |Две системные темы поддерживают шаблонные сопоставления. Примеры:

1. $SYS/broker/+/clients/#: Соответствует темам уведомлений о подключении и отключении клиента
2. $SYS/broker/+/clients/+/connected: Соответствует теме уведомления о подключении клиента
3. $SYS/broker/+/clients/+/disconnected: Соответствует теме уведомления о отключении клиента

4.9 Поддержка моста сообщений

Поддерживает промежуточные системы:

• Kafka

Функция моста сообщений позволяет легко интегрироваться с промежуточной системой.

1. mqttx.message-bridge.enable: Включение функции моста сообщений
2. mqttx.bridge-topics: Темы сообщений, которые требуется мостить, должны соответствовать требованиям topic для Kafka

При получении сообщения от клиента publish, сначала проверяется, включен ли мост сообщений, затем проверяются темы сообщений, и если они соответствуют темам для моста, сообщение публикуется на MQ.

Поддерживается односторонний мост: устройство(клиент) => mqttx => MQ

4.10 Поддержка ограничения скорости для тем

Используется алгоритм бакета токенов com.jun.mqttx.utils.RateLimiter для ограничения скорости для указанных тем.

Алгоритм бакета токенов см. здесь: https://stripe.com/blog/rate-limiters

Краткое объяснение концепции бакета токенов: имеется бакет с максимальной емкостью capacity, который заполняется токенами со скоростью replenish-rate. При каждом вызове интерфейса расходуется определенное количество токенов (token-consumed-per-acquire). Если токены доступны, запрос проходит.*Ограничение скорости применяется только к сообщениям с qos равным 0.

Пример конфигурации:

mqttx:
  rate-limiter:
    enable: true
    topic-rate-limits:
      # Пример 1
      - topic: "/test/a"
        capacity: 9
        replenish-rate: 4
        token-consumed-per-acquire: 3
      # Пример 2
      - topic: "/test/b"
        capacity: 5
        replenish-rate: 5
        token-consumed-per-acquire: 2

• capacity: емкость бака
• replenish-rate: скорость пополнения токенами
• token-consumed-per-acquire: количество токенов, потребляемых при каждом запросе

Формула расчета QPS:

1. Максимальное параллельное соединение: формула QPS = capacity ÷ token-consumed-per-acquire
   1. Пример 1: 9 ÷ 3 = 3
   2. Пример 2: 5 ÷ 2 = 2.5
2. Максимальное постоянное параллельное соединение: формула QPS = replenish-rate ÷ token-consumed-per-acquire
   1. Пример 1: 4 ÷ 3 ≈ 1.3
   2. Пример 2: 5 ÷ 2 = 2.5

4.11 Поддержка устойчивого хранения сообщений

Устойчивое хранение в mqttx зависит от redis. mqttx будет устойчиво хранить сообщения с параметрами cleanSession = false & qos > 0. Сообщения сериализуются объектом Serializer в массив байтов и затем сохраняются в redis.

На данный момент mqttx предоставляет два варианта сериализации:

1. JsonSerializer
2. KryoSerializer

По умолчанию используется JsonSerializer, чтобы обеспечить совместимость с предыдущими проектами; после версии v1.0.6.release KryoSerializer станет основным сериализатором.

Сериализацию можно изменить путем конфигурации параметра mqttx.serialize-strategy.#### 4.12 Поддержка базовой аутентификации

mqttx предоставляет базовый клиентский сервис аутентификации.

Параметры конфигурации:

1. mqttx.auth.url: адрес интерфейса для аутентификации.
2. mqttx.auth.timeout: время ожидания ответа от HttpClient.
3. mqttx.auth.is-mandatory: обязательность проверки имени пользователя и пароля.

Пользователи могут объявить параметр mqttx.auth.url в файле конфигурации. Объект com.jun.mqttx.service.impl.DefaultAuthenticationServiceImpl использует HttpClient для отправки POST-запроса на адрес mqttx.auth.url.

Запрос содержит данные mqtt пакета: username, password.

POST / HTTP/1.1
Host: mqttx.auth.url
Content-Type: application/json
Content-Length: 91

{
    "clientId": "device_id_test",
    "username": "mqttx",
    "password": "123456"
}

При успешной аутентификации ответ представляет собой JSON-строку:

{
    "authorizedSub": [
        "subTopic1",
        "subTopic2"
    ],
    "authorizedPub": [
        "pubTopic1",
        "pubTopic2"
    ]
}

Успешный ответ может использоваться вместе с 4.5 поддержка безопасности темы.

Примечание:

• Интерфейс возвращает http status = 200, что указывает на успешную аутентификацию, все остальные значения статуса считаются неудачной аутентификацией

5 Сообщение от разработчиков

1. Благодарим JetBrains за предоставленные лицензии для открытых проектов 2. Ветка с длительной поддержкой и обновлением

   1. v1.0: основанная на jdk8, при этом режим работы Redis IO — блокирующий.
   2. v1.2: основанная на jdk17, при этом режим работы Redis IO — неблокирующий.3. Чтобы сделать проект mqttx лучше, пожалуйста, активно сообщайте мне о вашем использовании и обучении этому проекту (откройте issue или присоединитесь к группе).

2. Будущие задачи

   • Разработка версии v1.0.8.RELEASE
   • Разработка версии v1.1.0.RELEASE
   • Разработка версии v1.2
   • Разработка версии v2.0
   • Устранение ошибок

3. Версия v1.2 была обновлена с JDK8 до JDK17

4. Ветка v2.0 будет начинаться с протокола MQTTv5

5. Рассмотрите использование протокола gossip для реализации функционала кластера, который больше не будет зависеть от Redis или Kafka

6. Поддержите автора чашкой шоколадного латте 😊

   

7. Общение в группе

   

6 Приложения

6.1 Настройки

В директории src/main/resources находятся три файла конфигураций:

1. application.yml
2. application-dev.yml
3. application-prod.yml

Два последних файла предназначены для разделения конфигураций в различных окружениях, что облегчает управление ими.Описание параметров конфигурации:

| mqttx.redis.pub-msg-set-prefix | mqttx:client:pubmsg: | префикс Redis set для клиентских сообщений pub; сохраняет pubmsg, удаляет после получения puback и pubrec | | mqttx.redis.pub-rel-msg-set-prefix | mqttx:client:pubrelmsg: | префикс Redis set для сообщений pubRel; сохраняет флаг pubrel сообщений, удаляет после получения pubcomp | | mqttx.redis.topic-set-key | mqttx:alltopic | ключ Redis set для всех тем; сохраняет все темы | | mqttx.redis.message-id-prefix | mqttx:messageId: | префикс messageId для клиентов без cleanSession; использует redis INCR команду | | mqttx.redis.client-topic-set-prefix | mqttx:client:topicset: | префикс Redis set для подписанных тем клиента; сохраняет все подписанные темы клиента | | mqttx.cluster.enable | false | включение режима работы в кластере | | mqttx.cluster.inner-cache-consistancy-key | mqttx:cache_consistency | ключ для проверки согласованности кэша при запуске приложения | | mqttx.cluster.type | redis | тип кластера | | mqttx.ssl.enable | false | включение SSL | | mqttx.ssl.client-auth | NONE | аутентификация сертификата клиента | | mqttx.ssl.key-store-location | classpath:tls/mqttx.keystore | расположение keyStore | | mqttx.ssl.trust-store-location | classpath:tls/mqttx.truststore| расположение trustStore |

Замечено, что последний столбец был прерван, поэтому продолжаем его:

| mqttx.ssl.key-store-password | password | пароль доступа к keyStore | | mqttx.ssl.trust-store-password | password | пароль доступа к trustStore |```markdown

topics`                     | `null`   | Список тем для моста между сообщениями                                                          |
|mqttx. rate-limiter. enable| `false`  | Включение ограничения скорости сообщений                                                        |
|mqttx. rate-limiter. token-rate-limit|          | Пример конфигурации, см.  [Поддержка ограничения скорости сообщений](#410-тема-ограничения-скорости-сообщений) |
|mqttx. auth. url| `null`   | Адрес интерфейса аутентификации MQTT conn username/password                                    |
|mqttx. auth. timeout| `3s` | Часовой лимит чтения                                                                           |
|mqttx. auth. is-mandatory| `false`  | Обязательность проверки имени пользователя и пароля в сообщении conn                            |
|mqttx. sharable-payload.|          |                                                                                                ||
payload-key-prefix| mqttx:sharable-payload:| префикс ключей Redis для хранения общего груза                 |
|mqttx. sharable-payload. unique-id-client-ids-set-prefix| mqttx:unique-id:client-ids:| список ассоциированных с общим грузом client_id                 |
|mqttx. sharable-payload. clean-work-interval| 1m| интервал времени между задачами очистки общего груза           |
|mqttx. sharable-payload. threshold-in-message| 128| пороговое значение для активации общего груза; если значение превышает порог, то происходит разделение груза |
```