Джо-mqtt-сервер

Автор (контакт для связи в случае вопросов)

Проект

Лёгкий сервер для интернета вещей, работающий по протоколу MQTT. Быстро развёртывается и поддерживает кластеризацию.

Архитектура

Построен на основе Java 8, Netty, Spring Boot, Redis и Hazelcast.

• Используется Netty для обеспечения коммуникации и анализа протокола.
• Spring Boot обеспечивает внедрение зависимостей и настройку свойств, что упрощает сборку и развёртывание.
• Поддерживаются протоколы SSL и WSS.
• Предоставляется возможность расширения функциональности.
  1. MemoryExtendProvider — не поддерживает обмен данными между узлами кластера, но поддерживает сообщения с QoS 0.
  2. RedisExtendProvider — поддерживает обмен данными между узлами кластера (используя механизм pub/sub от Redis), поддерживает все уровни QoS.
  3. HazelcastExtendProvider — поддерживает обмен данными между узлами кластера, поддерживает только сообщения с QoS 0.
  4. Если ни один из этих вариантов не подходит, можно создать собственное расширение, изменив конфигурацию mqtt.serverConfig.extendProviderClass=xxx.xxx.provider.XXXXProvider. Также можно добавить собственные переменные, следуя примеру конфигурации: пользовательские настройки расширения.

Не поддерживается

Не поддерживаются темы:

* пустые;
* начинающиеся или заканчивающиеся на '/';
* «joRootTopic»;
* содержащие только разделители, например, ab///c;
* содержащие запрещённые символы, например, ab/+11 или c/12#1.

Структура проекта

jo-mqtt
├── mqtt-broker — основная реализация функций MQTT-сервера;
├── mqtt-springboot — интеграция Springboot с MQTT;
└── mqtt-test — тестовые примеры для MQTT-сервера.

Функциональность

1. Реализация в соответствии со спецификацией MQTT 3.1.1.
2. Полная поддержка уровней качества обслуживания (QoS).
3. Поддержка сообщений о состоянии, сохранённых сообщений и повторной отправки сообщений.
4. Механизм сердцебиения.
5. Кластеризация.

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

• Локальное выполнение: клонируйте код на локальный компьютер и запустите программу MqttApplication. TCP-порт: 1883, TCP-SSL порт: 1888, WebSocket порт: 2883, WebSocket-SSL порт: 2888. Подробные настройки: application-local.properties.
• Запуск jar-файла (локальная среда): следуйте инструкциям в файле deploy/start.sh.

Ведение журнала

Используется фреймворк Logback.

Кластеризация

По умолчанию используется RedisExtendProvider для обмена данными между узлами. Обмен данными зависит от механизма pub/sub в Redis.

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

#server config
#tcp порт
#-1 означает, что порт не используется
mqtt.serverConfig.tcpPort=1883
#-1 означает, что порт не используется
mqtt.serverConfig.tcpSslPort=1888

#webSocket конфигурация
mqtt.serverConfig.webSocketPath=/joMqtt
#-1 означает, что порт не используется
mqtt.serverConfig.webSocketPort=2883
#-1 означает, что порт не используется
mqtt.serverConfig.webSocketSslPort=2888

mqtt.serverConfig.enableClientCA=false

mqtt.serverConfig.hostname=

#provider конфигурация, по умолчанию доступны следующие три реализации
#не поддерживает обмен данными между кластерами, не поддерживает сохранение сообщений
mqtt.serverConfig.extendProviderClass=joey.mqtt.broker.provider.MemoryExtendProvider

#поддерживает обмен данными между кластерами, поддерживает сохранение сообщений
#mqtt.serverConfig.extendProviderClass=joey.mqtt.broker.provider.RedisExtendProvider

#hazelcastProvider конфигурация, поддерживает обмен данными между кластерами, не поддерживает сохранение сообщений
#mqtt.serverConfig.extendProviderClass=joey.mqtt.broker.provider.HazelcastExtendProvider
#mqtt.customConfig.hazelcastConfigFile=classpath:hazelcast/hazelcast-local.xml
#mqtt.customConfig.hazelcastConfigFile=file:/home/hazelcast-local.xml

#пароль, используется шифрование sha256hex, пример содержит пароль в виде открытого текста и имя пользователя
mqtt.serverConfig.enableUserAuth=true
mqtt.serverConfig.authUsers[0].userName=local
mqtt.serverConfig.authUsers[0].password=25bf8e1a2393f1108d37029b3df5593236c755742ec93465bbafa9b290bddcf6
mqtt.serverConfig.authUsers[1].userName=admin
mqtt.serverConfig.authUsers[1].password=8c6976e5b5410415bde908bd4dee15dfb167a9c873fc4bb8a81f6f2ab448a918

#netty конфигурация
mqtt.nettyConfig.bossThreads=0
mqtt.nettyConfig.workerThreads=0
mqtt.nettyConfig.epoll=false
mqtt.nettyConfig.soBacklog=1024
mqtt.nettyConfig.soReuseAddress=true
mqtt.nettyConfig.tcpNoDelay=true
mqtt.nettyConfig.soSndBuf=65536
mqtt.nettyConfig.soRcvBuf=65536
mqtt.nettyConfig.soKeepAlive=true
mqtt.nettyConfig.channelTimeoutSeconds=200

#если используется RedisExtendProvider, необходимо настроить redisConfig
mqtt.customConfig.redisConfig.host=localhost
mqtt.customConfig.redisConfig.password=
mqtt.customConfig.redisConfig.port=6379
mqtt.customConfig.redisConfig.database=0
mqtt.customConfig.redisConfig.timeout=3000
mqtt.customConfig.redisConfig.pool.maxActive=50
mqtt.customConfig.redisConfig.pool.maxWait=1000
mqtt.customConfig.redisConfig.pool.maxIdle=50
mqtt.customConfig.redisConfig.pool.minIdle=20

#если включено ssl, необходимо настроить следующую информацию
mqtt.customConfig.sslContextConfig.sslKeyFilePath=ssl/jomqtt-server.pfx
mqtt.customConfig.sslContextConfig.sslKeyStoreType=PKCS12
mqtt.customConfig.sslContextConfig.sslManagerPwd=jo_mqtt
mqtt.customConfig.sslContextConfig.sslStorePwd=jo_mqtt

#настраиваемое имя узла, можно не настраивать, по умолчанию это UUID
#mqtt.customConfig.nodeName=jo_mqtt_1

#пользовательская конфигурация расширения
mqtt.customConfig.extConfig.k1=v1
mqtt.customConfig.extConfig.k2=v2
mqtt.customConfig.extConfig.k3.k31=v31
mqtt.customConfig.extConfig.k3.k32=v32

Расширение

Если текущая функциональность не удовлетворяет требованиям пользователя, можно расширить её самостоятельно. Для этого нужно унаследовать класс MemoryExtendProvider и переопределить соответствующие методы. Также можно добавлять собственные переменные, используя пример конфигурации: пользовательская настройка расширения. Чтобы изменить конфигурацию, используйте mqtt.serverConfig.extendProviderClass=joey.mqtt.broker.provider.XXXXProvider. 1. initMessageIdStore(); 2. Получение реализации хранилища сессий: ISessionStore initSessionStore(). 3. Получение реализации подписки на темы: ISubscriptionStore initSubscriptionStore(ISessionStore sessionStore). 4. Получение реализации хранения сообщений retain: IRetainMessageStore initRetainMessageStore(). 5. Получение реализации дублирования сообщений pub: IDupPubMessageStore initDupPubMessageStore(). 6. Получение реализации дублирования сообщений pubRel: IDupPubRelMessageStore initDupPubRelMessageStore(). 7. Получение реализации управления авторизацией: IAuth initAuthManager(List userList). 8. Получение реализации внутренней коммуникации кластера: IInnerTraffic initInnerTraffic(InnerPublishEventProcessor innerPublishEventProcessor). 9. Получение списка слушателей событий: List initEventListeners(). 10. Получение имени текущего узла экземпляра: String getNodeName(). 11. Инициализация sslContext: SslContext initSslContext(boolean enableClientCA) throws Exception.

Ссылки на реализацию

1. https://github.com/moquette-io/moquette.
2. https://gitee.com/recallcode/iot-mqtt-server.
3. https://github.com/Cicizz/jmqtt.
4. https://github.com/Wizzercn/MqttWk.
5. https://github.com/daoshenzzg/socket-mqtt.
6. https://github.com/1ssqq1lxr/iot_push.
7. https://mp.weixin.qq.com/s/9ErXfPUnJk20bPi5rSzcWA.
8. https://github.com/hivemq/hivemq-community-edition.
9. https://github.com/lets-mica/mica-mqtt aio tio实现.
10. https://github.com/quickmsg/smqtt.
11. https://github.com/jetlinks/jetlinks-community JetLinks — базовая платформа для интернета вещей.

Клиентские инструменты

1. https://mqttfx.jensd.de/index.php/download — загрузка клиентского инструмента.
2. https://github.com/emqx/MQTTX/releases — клиентский инструмент.
3. http://www.emqx.io/online-mqtt-client/#/recent_connections — онлайн-инструмент отладки.

Инструменты для нагрузочного тестирования

1. https://github.com/takanorig/mqtt-bench — скопируйте файл mqtt-mock из папки mqtt-test в тестовую машину и запустите его.
2. https://github.com/daoshenzzg/mqtt-mock — выполните команду ./mqtt-mock -broker "tcp://localhost:1883" -c 6000 -n 1000 -action sub -topic test -username local -password local -debug true.

Рекомендуемые инструменты

1. https://github.com/looly/hutool — Hutool, небольшая и полная библиотека классов инструментов Java. Автор отвечает на вопросы, настоятельно рекомендуется.
2. http://www.tongxinmao.com/txm/webmqtt.php — онлайн-тестирование MQTT.
3. https://www.emqx.io/docs/zh/v4.3/tutorial/tune.html — настройка системы.
4. https://blog.csdn.net/Just_shunjian/article/details/78288229 — оптимизация ядра Linux — увеличение максимального количества подключений TCP.
5. https://github.com/xiaojiaqi/C1000kPracticeGuide — руководство по оптимизации C1000k.