# talkweb_mqtt

**Repository Path**: tw-dream/talkweb_mqtt

## Basic Information

- **Project Name**: talkweb_mqtt
- **Description**: MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的轻量级协议，该协议构建于TCP/IP协议之上，MQTT最大优点在于可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。
- **Primary Language**: Unknown
- **License**: Apache-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 11
- **Created**: 2022-01-27
- **Last Updated**: 2022-01-27

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# TW-MQTT broker

#### 介绍
MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的轻量级协议，该协议构建于TCP/IP协议之上，MQTT最大优点在于可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。

#### 源码目录

```
tw-mqtt
├── auth
├── cluster
├── config
├── enums
├── kafka
├── model
├── service
├── util
├── MqttBrokerApplication.java

auth: mqtt 连接认证、订阅认证，
cluster: redis 存储集群节点、遗嘱消息、保留消息。
config: Spring boot 自动配置。
enums: redis 缓存 key 枚举。
kafka: mqtt 集群内消息交换和 mqtt 上下行消息转发。
model: redis 存储模型。
service: 自定义服务，可用于二次开发。
util: 工具类
```

#### 系统架构

![架构图](docs/images/iot-diagram.png)

#### 依赖项
- redis 遗嘱和保留消息存储
- kafka 集群消息流转和上下行消息通道

#### 功能
- [x] 基于 kafka 轻松实现千万级设备连接的 Mqtt Broker，
- [x] 支持完整的 MQTT v3.1、v3.1.1 和 v5.0 协议接入。
- [x] 支持 websocket mqtt 子协议（支持 mqtt.js）接入。
- [x] 支持 http rest api 调用，[http api 文档详见](docs/http-api.md)。
- [x] 支持 MQTT 遗嘱消息、保留消息。
- [x] 支持对接 Prometheus + Grafana 实现监控。

#### 默认端口
| 端口号 | 协议            | 说明                             |
| ------ | --------------- | -------------------------------- |
| 1883   | tcp             | mqtt tcp 端口                    |
| 8083   | http、websocket | http api 和 websocket mqtt 子协议端口 |

#### 默认连接账号
| 属性        | 默认值 |
| ---------- | ---------- |
| username   | twmqtt |
| password   | twmqtt |

#### 二开说明
1、 完善 auth 包下的 MqttAuthHandler、MqttSubscribeValidator 连接和订阅认证实现。

2、 监听 Kafka 上行消息和发送下行消息到 Kafka，业务服务 kafka 消息序列化反序列化参考源码 kafka 目录下的 [MessageDeserializer](src/main/java/com/talkweb/iot/mqtt/broker/kafka/MessageDeserializer.java)、[MessageSerializer](src/main/java/com/talkweb/iot/mqtt/broker/kafka/MessageSerializer.java)。

**Redis key列表**

| Redis key                       | 类型   | 内容          | 说明                      |
| ------------------------------- | ------ | ------------- | ------------------------- |
| mqtt:server:nodes               | Set    | 集群 nodeName | mqtt 服务端节点           |
| mqtt:connect:status:${nodeName} | Set    | clientId 集合 | mqtt 集群下客户端连接明细 |
| mqtt:messages:will:${clientId}  | String | 遗嘱消息      | 遗嘱消息存储              |
| mqtt:messages:retain:${topic}   | String | 保留消息      | 保留消息存储              |

**Kafka topic列表**

| topic                           | 说明                                                    |
| ------------------------------- | ------------------------------------------------------- |
| mqtt.bridge.message.up          | 设备 -> 云端 上行消息 （业务服务监听设备消息的topic）   |
| mqtt.bridge.message.down        | 云端 -> 设备 下行消息 （业务服务发送消息到设备的topic） |
| mqtt.bridge.client.connected    | 设备连接通知                                            |
| mqtt.bridge.client.disconnected | 设备断开链接通知                                        |

#### 安装教程

1. 本地安装 java 和 maven 工具，未来会直接打好 Docker 镜像推送到 docker hub 官方。

2. 代码根目录下执行 `mvn package` 构建。
   
3. 执行 `docker build -t "talkweb/mqtt" .` 构建 docker 镜像。

4. 执行 `docker run -d --name=mqtt-broker -e spring.redis.host=127.0.0.1 -e spring.kafka.bootstrap-servers=127.0.0.1:9092 talkweb/mqtt` 运行。

5. mqtt broker 参数配置详见[mqtt 服务配置文档](docs/config.md)。

#### mqtt 客户端工具

- [mqttx 开发调试工具](https://mqttx.app/cn/)
- [浏览器、小程序 mqtt.js 接入](https://github.com/mqttjs/MQTT.js)
- [mqtt client c 接入](https://github.com/eclipse/paho.mqtt.c)
- [mqtt client Android 接入](https://github.com/eclipse/paho.mqtt.android)
- [mqtt client ios 接入](https://github.com/novastone-media/MQTT-Client-Framework)
- [OpenHarmony的终端设备SDK，快速接入拓维IOT物联网平台](https://gitee.com/talkweb-iioh/TLlink-sdk-oh)

#### 参与贡献

1.  Fork 本仓库
2.  新建 Feat_xxx 分支
3.  提交代码
4.  新建 Pull Request