登陆

通过 MQTT 学习协议制定与实现(二)

admin 2022-11-24 6人围观 ,发现0个评论

开启掘金成长之旅!这是我参与「掘金日新计划 · 12 月更文挑战」的第3天,点击查看活动详情

写在前面

这是 MQTT 协议学习的第二篇,上一篇通过 MQTT 学习协议制定与实现(一)介绍了 MQTT 协议的报文格式。

Fixed header 固定报头,所有控制报文都包含
Variable header 可变报头,部分控制报文包含
Payload 有效载荷,部分控制报文包含

本篇通过 Netty 简单实现一个 MQTT Server,利用 MQTTX 客户端发送控制报文,通过 Wiresharks 抓包来分析不同类型的控制报文,固定报头、可变报头、有效载荷部分。

简单 MQTT Server 实现

Netty MQTT Server 定义

定义一个 Netty MQTT Server,通过 ServerBootstrap 实现,关于 Netty 我想后面也系统性的写文章介绍,对于不了解 Netty 的同学可以将 Netty 先看成黑盒,就简单的理解成起了一个 Server 绑定 60000端口。

这里重点关注初始化 channel 在 channel pipeline 中添加 MqttEncoder.INSTANCEMqttDecodermqttHander,对网络通道中传输的 MQTT 字节流进行编码、解码以及业务逻辑处理。

@Component @RequiredArgsConstructor public class MqttAgent { ​     private final MqttHandler mqttHandler;     private EventLoopGroup _bossGroup;     private EventLoopGroup _workGroup; ​     @PostConstruct     public void start() throws Exception {         try {             _bossGroup = new NioEventLoopGroup(1);             _workGroup = new NioEventLoopGroup();             ServerBootstrap serverBootstrap = new ServerBootstrap();             serverBootstrap.group(_bossGroup, _workGroup)                     .channel(NioServerSocketChannel.class)                     .childHandler(new ChannelInitializer<SocketChannel>() {                         @Override                         protected void initChannel(SocketChannel ch) {                             ch.pipeline().addLast("mqttEncoder", MqttEncoder.INSTANCE);                             ch.pipeline().addLast("mqttDecoder", new MqttDecoder());                             ch.pipeline().addLast("mqttHandler", mqttHandler);                         }                     });             serverBootstrap.bind(                     "0.0.0.0",                     60000             ).sync();         } catch (Exception e) {             try {                 _bossGroup.shutdownGracefully();                 _workGroup.shutdownGracefully();             } catch (Exception shutdownException) {                 // ignore             }             // rethrow exception while agent does not start up normally.             throw e;         }     } } 复制代码

MQTT Handler 实现

MqttHandler 主要处理 Mqtt 报文的处理逻辑,如连接建立时对用户账号密码等权限检查,这里做最简单的处理,打印出 Mqtt 报文然后回复一个 MqttConnAckMessage,代码如下:

@Component @Slf4j @ChannelHandler.Sharable public class MqttHandler extends SimpleChannelInboundHandler<MqttMessage> {          @Override     protected void channelRead0(ChannelHandlerContext ctx, MqttMessage mqttMessage) throws Exception {         //打印mqtt 消息         log.info("mqtt message:{}", mqttMessage);         //回复 connect ack 控制报文         ctx.channel().writeAndFlush(connAckMessageInstance(MqttConnectReturnCode.CONNECTION_ACCEPTED));     } ​     //构造一个 MqttConnAckMessage     public static MqttConnAckMessage connAckMessageInstance(MqttConnectReturnCode returnCode) {         //连接确认标志+连接返回码(2位)         MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false,                 MqttQoS.AT_MOST_ONCE, false, 2);         MqttConnAckVariableHeader variableHeader = new MqttConnAckVariableHeader(returnCode, false);         return new MqttConnAckMessage(fixedHeader, variableHeader);     } } 复制代码

MqttConnAckMessage 报文只有固定报头和可变报头,没有有效载荷部分。

  • 固定报头:控制报文类型值为2,标志位0,剩余长度为两个字节的可变报头
  • 可变报头:2个字节,一个字节连接确认标志,第二个字节是连接返回码

image-20221124222213855.png

验证程序

使用 MQTTX 客户端,编辑连接信息,mqtt server host:127.0.0.1, port:60000,启动服务端。

  • 建立连接 image-20221124222558283.png
  • 连接成功,发送消息 image-20221124222929921.png

服务端成功接收到消息

2022-11-24 22:39:49.304  INFO 8299 --- [ntLoopGroup-3-1] com.demo.mqttdemo.MqttHandler: mqtt message:MqttFixedHeader[messageType=PUBLISH, isDup=false, qosLevel=AT_MOST_ONCE, isRetain=false, remainingLength=22]  复制代码

Wiresharks 抓包

打开 Wiresharks 客户端,MQTTX 发送连接请求,查看 Wiresharks 抓包:如图所示 No61,No62,No63 则是大名鼎鼎的三次握手,No67,No69 则是 MQTT Connect 控制报文与 ConnAck 控制报文。

image-20221124225713907.png

Connect 控制报文

image-20221124230044717.png

由报文 DATA 部分可以看出

固定报头两字节:10 1a,10为 Connect 控制报文类型,1a则为剩余长度

CONNECT 报文的可变报头:按下列次序包含四个字段:协议名(Protocol Name),协议级别(Protocol Level),连接标志(Connect Flags)和保持连接(Keep Alive)。

00 04 4d 51 54 54 这固定的6个字节就表明了 MQTT 协议名

说明76543210
协议名
byte 1长度 MSB (0)00000000
byte 2长度 LSB (4)00000100
byte 3‘M’01001101
byte 4‘Q’01010001
byte 5‘T’01010100
Byte 6‘T’01010100

ConnAck 控制报文

image-20221124231203686.png

固定报头:20 02 两个字节表示控制报文类型以及剩余长度,ConnAck 固定报头格式为:

Bit76543210
byte 100100000
byte 200000010

可变报头:00 00 两个字节表示连接确认标志与连接返回码,连接已接收返回码则是 0x00 所以两个字节为 00 00

请发表您的评论
请关注微信公众号
微信二维码
不容错过
Powered By Z-BlogPHP