IOT云平台 simple（4）springboot netty实现简单的mqtt broker

水深无声 2024-04-01 19:36 74阅读 0赞

本系列教程包括：  
[IOT云平台 simple（0）IOT云平台简介][IOT_ simple_0_IOT]  
[IOT云平台 simple（1）netty入门][IOT_ simple_1_netty]  
[IOT云平台 simple（2）springboot入门][IOT_ simple_2_springboot]  
[IOT云平台 simple（3）springboot netty实现TCP Server][IOT_ simple_3_springboot netty_TCP Server]  
[IOT云平台 simple（4）springboot netty实现简单的mqtt broker][IOT_ simple_4_springboot netty_mqtt broker]  
[IOT云平台 simple（5）springboot netty实现modbus TCP Master][IOT_ simple_5_springboot netty_modbus TCP Master]  
[IOT云平台 simple（6）springboot netty实现IOT云平台基本的架构（mqtt、Rabbitmq）][IOT_ simple_6_springboot netty_IOT_mqtt_Rabbitmq]

常见的开源mqttBroker很多，如：  
Mosquitto、emqx；  
这里简单的介绍了mqtt，然后利用springboot netty实现了简单的mqtt Broker。

mqtt Broker：springboot netty实现；  
mqtt client：MQTT.fx工具软件；

## 1 开发 ##

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的"轻量级"通讯协议。  
![在这里插入图片描述][d904db399b9a430984a9a4830d29a2e5.png]  
mqtt是运行在TCP/IP之上的应用层，所以mqtt Broker基本就是个TCP Server。  
创建主要的类：  
1） MqttBrokerChannelHandler：  
server channel处理的类；  
2 ）MqttBrokerChannelInitializer  
server channel初始化的类  
3）MqttBroker  
server类。  
4）MqttClientStartListener：监听到springboot启动后，启动MqttBroker。

mqtt数据协议：  
数据结构包括：  
1）固定报头（Fixed header）；  
2）可变报头（Variable header）；  
3）有效载荷（Payload）。  
![在这里插入图片描述][7dfde1f3b8274244904985b0b3b6be57.png]

其中，MqttBrokerChannelHandler类中实现了mqtt数据的解析处理。

@Override
        public void channelRead(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
        
            MqttMessage mqttMessage = (MqttMessage) msg;
            log.info("--------------------------begin---------------------------*");
            log.info("来自终端:" + channelHandlerContext.channel().remoteAddress());
            log.info("接收消息：" + mqttMessage.toString());
            try {
        
                MqttMessageType type = mqttMessage.fixedHeader().messageType();
                MessageStrategy messageStrategy =  messageStrategyManager.getMessageStrategy(type);
                if(messageStrategy!=null){
        
                    messageStrategy.sendResponseMessage(channelHandlerContext.channel(),mqttMessage);
                }
            }catch (Exception e) {
        
                e.printStackTrace();
            }
            log.info("--------------------------end---------------------------*");
        }

在这里我们定义一个策略接口，实现对mqtt 消息的解析返回。

public interface MessageStrategy {
        
        void sendResponseMessage(Channel channel, MqttMessage mqttMessage);
    }

然后我们就可以实现不同类型消息的策略处理，如对Connect消息的处理：

@Slf4j
    public class ConnectAckMessageStrategy implements MessageStrategy{
        
    
        @Override
        public void sendResponseMessage(Channel channel, MqttMessage mqttMessage) {
        
            MqttConnectMessage mqttConnectMessage = (MqttConnectMessage)mqttMessage;
    
            /*---------------------------解析接收的消息----------------------------*/
            MqttFixedHeader mqttFixedHeader = mqttConnectMessage.fixedHeader();
            MqttConnectVariableHeader mqttConnectVariableHeader = mqttConnectMessage.variableHeader();
    
            /*---------------------------构建返回的消息---------------------------*/
            //	构建返回报文， 固定报头
            MqttConnAckVariableHeader mqttConnAckVariableHeader =new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, mqttConnectVariableHeader.isCleanSession());
            MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.CONNACK,mqttFixedHeader.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeader.isRetain(), 0x02);
    
            //	构建CONNACK消息体
            MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeaderBack, mqttConnAckVariableHeader);
            log.info("返回消息:"+connAck.toString());
            channel.writeAndFlush(connAck);
        }
    }

最后我们定义了MessageStrategyManager类实现对不同类型消息的策略处理：

@Slf4j
    @Component
    public class MessageStrategyManager {
        
        public Map<MqttMessageType, MessageStrategy> messageStrategyMap = new HashMap<>();
    
        //根据消息类型获取对应的策略类
        public  MessageStrategy getMessageStrategy(MqttMessageType messageType){
        
            switch (messageType){
        
                case CONNECT:
                    return new ConnectAckMessageStrategy();
                case PUBLISH:
                    return new PublishAckMessageStrategy();
                case PUBREL:
                    return new PublishCompleteMessageStrategy();
                case SUBSCRIBE:
                    return new SubscribeAckMessageStrategy();
                case UNSUBSCRIBE:
                    return new UnSubscribeAckMessageStrategy();
                case PINGREQ:
                    return new PingMessageStrategy();
                default:
                    return null;
            }
        }
    
        //根据消息类型获取返回消息的类型
        private  static MqttMessageType getResMqttMessageType(MqttMessageType messageType){
        
            switch (messageType){
        
                case CONNECT:
                    return MqttMessageType.CONNACK;
                case PUBLISH:
                    return MqttMessageType.PUBACK;
                case PUBREL:
                    return MqttMessageType.PUBLISH;
                case SUBSCRIBE:
                    return MqttMessageType.SUBACK;
                case UNSUBSCRIBE:
                    return MqttMessageType.UNSUBACK;
                case PINGREQ:
                    return MqttMessageType.PINGRESP;
                default:
                    return null;
            }
        }
    }

## 2 验证 ##

### 2.1 MQTT.fx连接broker ###

MQTT.fx发送消息的类型：CONNECT  
broker返回消息的类型：CONNACK  
![在这里插入图片描述][4cd1de433c4e44d096edc19d548a1909.png]

### 2.2 MQTT.fx向broker发送心跳 ###

MQTT.fx发送消息的类型：PINGREQ  
broker返回消息的类型：PINGRESP  
![在这里插入图片描述][646bd54f5b7e418e8bb835d969ccdf43.png]

### 2.3 MQTT.fx向broker发布topic ###

首先MQTT.fx发布topic为：

weather/beijing

发布内容为：

The low temperature is 20C and the high temperature is 25 C

![在这里插入图片描述][dca65aa80a504239a4a2de00a39aeaa8.png]  
从log看到broker已经收到topic：  
MQTT.fx发送消息的类型：PUBLISH  
这里qosLevel=AT\_MOST\_ONCE。  
![在这里插入图片描述][03058375b121432b9ee59ea7522e6b44.png]

### 2.4 MQTT.fx断开连接broker ###

MQTT.fx发送消息的类型：DISCONNECT  
![在这里插入图片描述][d3234b305196470ea63db9d84a952f4d.png]

代码详见：  
[https://gitee.com/linghufeixia/iot-simple][https_gitee.com_linghufeixia_iot-simple]  
code3

[IOT_ simple_0_IOT]: https://blog.csdn.net/afei8080/article/details/115539444
[IOT_ simple_1_netty]: https://blog.csdn.net/afei8080/article/details/115539643
[IOT_ simple_2_springboot]: https://blog.csdn.net/afei8080/article/details/115540777
[IOT_ simple_3_springboot netty_TCP Server]: https://blog.csdn.net/afei8080/article/details/127832895
[IOT_ simple_4_springboot netty_mqtt broker]: https://blog.csdn.net/afei8080/article/details/127899022
[IOT_ simple_5_springboot netty_modbus TCP Master]: https://blog.csdn.net/afei8080/article/details/128242813
[IOT_ simple_6_springboot netty_IOT_mqtt_Rabbitmq]: https://blog.csdn.net/afei8080/article/details/128557674
[d904db399b9a430984a9a4830d29a2e5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/cd6aaa0f205d4379b4e3d16f2951587e.png
[7dfde1f3b8274244904985b0b3b6be57.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/1cbee16d82f74520955465f476e1f37e.png
[4cd1de433c4e44d096edc19d548a1909.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/4b94674b6f494a43a5aaea1a86640a43.png
[646bd54f5b7e418e8bb835d969ccdf43.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/6101be5012034c1f94c478842abaaff0.png
[dca65aa80a504239a4a2de00a39aeaa8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/514e6c1292dc498fa3090c3577239bd0.png
[03058375b121432b9ee59ea7522e6b44.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/17650cfd63b44f738e2752bec8f010a6.png
[d3234b305196470ea63db9d84a952f4d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/01/45021843d4534c4e8175c804669aef8a.png
[https_gitee.com_linghufeixia_iot-simple]: https://gitee.com/linghufeixia/iot-simple