Проект MQTTX

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

| поле                  | Описание                                                                 |
| ---------------------- | ------------------------------------------------------------------------- |
| `активное_количество_соединений` | Текущее количество активных соединений                                 |
| `временная_марка`     | Временная метка; (`гггг-мм-дд чч:мм:сс`)                                  |
| `версия`             | Версия `mqttx`                                                            |

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

Поддержка промежуточного программного обеспечения для сообщений:

- [x] kafka

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

1. `mqttx.message_bridge_enable`: Включение функции моста сообщений
2. `mqttx.bridge_topics`: Темы, которым требуется мостовое соединение

После того как `mqttx` получает сообщение от клиента **публикации**, он сначала проверяет, включен ли мостовой режим, затем проверяет, является ли тема темой, которая требует мостового соединения, и наконец публикует сообщение в **MQ**.

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

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

1. В состоянии кластера следует рассмотреть возможность интеграции регистрации сервиса, что позволит удобнее управлять состоянием кластера. Можно использовать `consul`. Возможно, я создам ветку для реализации этого.   > На самом деле я хотел бы внедрить `SpringCloud`, но чувствую, что `springcloud` немного тяжеловат, поэтому возможно, я создам ветку для его реализации.

2. Устранение ошибок и оптимизация продолжаются, но в основном зависит от студентов, использующих и изучающих `mqttx`, чтобы они предоставляли мне обратную связь по проблемам (если нет обратной связи, я буду считать это отсутствием ~).   > Это действительно важно, но пока мало студентов обращаются ко мне за обратной связью. Я человек ограниченной власти после окончания всего.3. Управляемая платформа [mqttx-admin](https://github.com/Amazingwujun/mqttx-admin), основанная на `vue2.0`, `element-ui`, в настоящее время находится в процессе разработки. Обновление функциональности `mqttx` будет приостановлено на некоторое время ~~(Последнее время изучал [mqtt5](http://docs.oasis-open.org/mqtt/mqtt/v5.0/csprd02/mqtt-v5.0-csprd02.html))~~. В ходе разработки проекта было выявлено, что требуются некоторые изменения в `mqttx`, но эти изменения не следует отправлять в основной ветвь mqttx (например, аутентификация безопасности темы должна взаимодействовать с `mqttx-platform`, возможно, я буду использовать [Retrofit](https://square.github.io/retrofit/) для обработки вызовов API, фактически можно использовать `feign`, я считаю, что эти два подхода схожи), мне следует открыть бизнес-ветку для обработки этих изменений. Кстати, разработка проектов на `javascript` так крута, почему ты не подумал об этом раньше?   > Вначале мне пришлось направить некоторые усилия на производное проект `mqttx-admin`, но позднее выяснилось, что в `mqttx` всё ещё остаётся слишком много задач, и пришлось изменить план.4. [Оценка производительности](#63-оценка-производительности) показывает, что производительность `mqttx` может быть улучшена. Я буду модифицировать логику обработки `pub/sub` в версии `v1.1.0.RELEASE`.

   > Основной акцент — это замена `StringRedisTemplate` на `ReactiveStringRedisTemplate`, переход от **синхронного** режима к **асинхронному**.

5. Введение в направление развития

   `mqttx` создаёт две ветви:

    - v1.0: `com.jun.mqttx.service.impl` синхронный интерфейс
    - v1.1: `com.jun.mqttx.service.impl` изменён на асинхронный интерфейс

   Поддержка [mqtt5](http://docs.oasis-open.org/mqtt/mqtt/v5.0/csprd02/mqtt-v5.0-csprd02.html) начинается с версии `v1.0`, поэтому неудивительно, что следующая версия — это `v1.0.6.RELEASE`.

6. Группа обмена

<img src="https://s1.ax1x.com/2020/10/10/0ytoSx.jpg" alt="QR-код группы" height="300" />

## 6 Расписание

### 6.1 Настройка параметров конфигурации

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

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

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

| Конфигурация            | По умолчанию | Описание                                                                                   |
|-------------------------|--------------|--------------------------------------------------------------------------------------------|
| `mqttx.version`         | Из `pom.xml` | Версия                                                                                      |
| `mqttx.brokerId`        | `1`          | Уникальный идентификатор приложения                                                         |
| `mqttx.heartbeat`       | `60s`        | Изначальное значение heartbeat будет перезапрошено сообщением keepalive в коннект-сообщении |
| `mqttx.host`            | `0.0.0.0`    | Адрес прослушивания                                                                         |
| `mqttx.soBacklog`       | `512`        | Очередь обработки TCP соединений                                                            |
| `mqttx.enableTopicSubPubSecure` | `false` | Функция безопасности подписки/публикации темы клиентами, когда активирована, она ограничивает темы, которые могут быть опубликованы или подписаны клиентами |
| `mqttx.enableInnerCache` | `true`       | После публикации каждого сообщения требуется запрос к Redis для получения списка подписанных клиентов. При активации этой функции будет создано отображение отношений между темой и клиентом в оперативной памяти, что позволит приложению непосредственно обращаться к данным в памяти |
| `mqttx.enableTestMode`   | `false`      | Переключатель тестового режима, система переходит в тестовый режим после его активации     |
| `mqttx.redis.clusterSessionHashKey` | `mqttx.session.key` | Ключ отображения Redis; ключ хранения сессий для кластера                    |
| `mqttx.redis.topicPrefix` | `mqttx:topic:` | Префикс темы; сохранение отношения между темой и клиентом                               |
| `mqttx.redis.retainMessagePrefix` | `mqttx:retain:` | Префикс сохраняемого сообщения; сохранение сообщения retain                       |

|mqttx.redis.|```markdown
| Настройка                                             | Значение                          | Описание                                                                                                                                                                                                                   |
|-------------------------------------------------------|-----------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `mqttx.redis.pubMsgSetPrefix`                         | `mqttx:client:pubmsg:`            | Префикс набора Redis для сообщений публикации клиента; сохранение сообщения pubmsg, удаление после получения puback для получения pubrec                                                                                                                                         |
| `mqttx.redis.pubRelMsgSetPrefix`                      | `mqttx:client:pubrelmsg:`         | Префикс набора Redis для сообщений pubRel клиента; сохранение сообщения pubrel, удаление после получения сообщения pubcom                                                                                                                                                    |
| `mqttx.redis.topicSetKey`                             | `mqttx:alltopic`                  | Набор всех тем; ключ значения Redis; сохранение всех тем                                                                                                                                                                                                                           |
| `mqttx.cluster.enable`                                | `false`                           | Переключатель кластера                                                                                                                                                                                                                                                               |
| `mqttx.cluster.innerCacheConsistancyKey`              | `mqttx:cache_consistence`        | После запуска приложения первым делом выполняется запрос к Redis без этого значения ключа, затем проверяется согласованность                                                                                                                                                     |
| `mqttx.ssl.enable`                                    | `false`                           | Переключатель SSL                                                                                                                                                                                                                                                                     |
| `mqttx.ssl.client-auth`                               | `false`                           | Проверка сертификата клиента                                                                                                                                                                                                                                                         |
| `mqttx.ssl.keyStoreLocation`                          | `classpath:tls/mqttx.keystore`   | Расположение keyStore                                                                                                                                                                                                                                                                |
| `mqttx.ssl.keyStorePassword`                          | `123456`                          | Пароль keyStore                                                                                                                                                                                                                                                                       |
| `mqttx.ssl.keyStoreType`                              | `pkcs12`                          | Тип keyStore                                                                                                                                                                                                                                                                           |
| `mqttx.socket.enabled`                                | `true`                            | Переключатель сокета                                                                                                                                                                                                                                                                  |
| `mqttx.socket.port`                                   | `1883`                           | Порт прослушивания сокета                                                                                                                                                                                                                                                            |
| `mqttx.websocket.enable`                              | `false`                          | Переключатель WebSocket                                                                                                                                                                                                                                                               |
| `mqttx.websocket.port`                                | `8083`                           | Порт прослушивания WebSocket                                                                                                                                                                                                                                                          |
| `mqttx.websocket.path`                                | `/mqtt`                          | Путь WebSocket                                                                                                                                                                                                                                                                         |
| `mqttx.share-topic.enable`                            | `true`                           | Переключатель функции общего топика                                                                                                                                                                                                                                                   |
| `mqttx.share-topic.share-sub-strategy`                | `round-robin`                    | Стратегия балансировки нагрузки, в настоящее время поддерживаются случайный, опрос, хэширование                                                                                                                                                                                 |
| `mqttx.sys-topic.enable`                              | `false`                          | Переключатель функции системного топика                                                                                                                                                                                                                                               |
``````diff
+ interval` | `60c` | интервал периодического запуска |
+ | `mqttx.sys-topic.qos` | `0` | уровень качества обслуживания топика |
+ | `mqttx.message-bridge.enable` | `false` | переключатель функции моста сообщений |
+ | `mqttx.message-bridge.topics` | `null` | список тем, которым требуется мостовое соединение |

C:\Users\Jun\go\windows_amd64>mqtt-bench.exe -broker=tcp://localhost:1883 -action=pub -clients=1000 -qos=0 -count=1000
2020-09-30 15:33:54.462089 +0800 CST Начало бенчмарка
2020-09-30 15:34:33.6010217 +0800 CST Конец бенчмарка

Результат: брокер=tcp://localhost:1883, клиентов=1000, total_count=1000000, длительность=39134мс, пропускная_способность=25553.23сообщений/сек

C:\Users\Jun\go\windows_amd64>mqtt-bench.exe -broker=tcp://localhost:1883 -action=pub -clients=1000 -qos=1 -count=1000
2020-09-30 15:29:17.9027515 +0800 CST Начало бенчмарка
2020-09-30 15:30:25.0316915 +0800 CST Конец бенчмарка

Результат: брокер=tcp://localhost:1883, клиентов=1000, total_count=1000000, длительность=67124мс, пропускная_способность=14897.80сообщений/сек

C:\Users\Jun\go\windows_amd64>mqtt-bench.exe -broker=tcp://localhost:1883 -action=pub -clients=1000 -qos=2 -count=1000
2020-09-30 15:37:00.0678207 +0800 CST Начало тестирования
2020-09-30 15:38:55.4419847 +0800 CST Конец тестирования

Результат: брокер=tcp://localhost:1883, клиентов=1000, total_count=1000000, длительность=115369мс, пропускная_способность=8667.84сообщений/сек

C:\Users\Jun\go\windows_amd64>mqtt-bench.exe -broker=tcp://localhost:1883 -action=pub -clients=1000 -qos=0 -count=1000
2020-09-30 17:03:55.7560928 +0800 CST Начало тестирования
2020-09-30 17:04:36.2080909 +0800 CST Конец тестирования

Результат: брокер=tcp://localhost:1883, клиентов=1000, total_count=1000000, длительность=40447 мс, пропускная способность=24723.71 сообщений/сек

C:\Users\Jun\go\windows_amd64>mqtt-bench.exe -broker=tcp://localhost:1883 -action=pub -clients=1000 -qos=1 -count=1000
2020-09-30 17:06:18.9136484 +0800 CST Начало тестирования
2020-09-30 17:08:20.9072865 +0800 CST Конец тестирования

Результат: брокер=tcp://localhost:1883, клиентов=1000, total_count=1000000, длительность=121992 мс, пропускная способность=8197.26 сообщений/сек

C:\Users\Jun\go\windows_amd64>mqtt-bench.exe -broker=tcp://localhost:1883 -action=pub -clients=1000 -qos=2 -count=1000
2020-09-30 17:09:35.1314262 +0800 CST Начало тестирования
2020-09-30 17:13:10.7914125 +0800 CST Конец тестирования
```Результат: брокер=tcp://localhost:1883, клиентов=1000, total_count=1000000, длительность=215656 мс, пропускная способность=4637.01 сообщений/сек
```| Количество одновременных соединений | Действие | Размер одного сообщения | Количество сообщений в одном соединении | Общее количество сообщений | QoS | Время выполнения | QPS |
| ------------ | -------- | ------------ | -------------- | -------- | ---- | -------- | ------- |
| `1000` | Отправка сообщения | `1024 байта` | `1000` | Один миллион | `0` | `40.4 секунды` | `24723` |
| `1000` | Отправка сообщения | `1024 байта` | `1000` | Один миллион | `1` | `121.9 секунды` | `8197` |
| `1000` | Отправка сообщения | `1024 байта` | `1000` | Один миллион | `2` | `215.6 секунд` | `4637` |```**Расход ресурсов: `cpu: 45%`, `mem: 440 MB`**