MQTT协议代码实现详解一

为了打发空闲时间会对之前写的MQTT协议的代码实现进行详细讲解；主要是讲解的内容是对MQTT网络通讯协议的分析和实现，具体涉及对网络数据流拆分处理，因此很多内容都涉及到具体的代码功能，如果你有兴趣学习这方面的那应该是有一定帮助的。接来就讲解BeetleX.MQTT.Protocols的设计和具体实现细节（对于MQTT协议是5.0版本）。
参考文档 中文：vitsumoc.github.io/ mqtt-v5-0-chinese.ht ml
英文: d ocs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html
编程语言: C#
完整代码: //github.com/beetlex-io/mqtt

什么是MQTT
M QTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的「轻量级」通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布(最新版本协议于2019年定稿的5.0)。

基础类型
通讯协议是对数据流规范的定义，根据具体规范进行数据拆分和读取。每个被拆分成一个完整消息的可以称作为消息包，而消息包则有多个数据成员组件，而这些数据成员则是协议的基础类型。而在MQTT协议操作过程中涉及的类型如下:

bool tag = (0b0000_0001 & data) > 0;//读取0位是否为1data|=0b0000_0001//设置第0位为1

public classInt7bit{ [ThreadStatic]privatestatic Byte[] mDataBuffer;publicvoidWrite(System.IO.Stream stream, intvalue) {if (mDataBuffer == null) mDataBuffer = newbyte[32];var count = 0;var num = (UInt64)value;while (num >= 0x80) { mDataBuffer[count++] = (Byte)(num | 0x80); num >>= 7; } mDataBuffer[count++] = (Byte)num; stream.Write(mDataBuffer, 0, count); }privateuint mResult = 0;privatebyte mBits = 0;publicint? Read(System.IO.Stream stream) { Byte b;while (true) {if (stream.Length < 1)returnnull;var bt = stream.ReadByte();if (bt < 0) { mBits = 0; mResult = 0;thrownew BXException("Read 7bit int error:byte value cannot be less than zero!"); } b = (Byte)bt; mResult |= (UInt32)((b & 0x7f) << mBits);if ((b & 0x80) == 0) break; mBits += 7;if (mBits >= 32) { mBits = 0; mResult = 0;thrownew BXException("Read 7bit int error:out of maximum value!"); } } mBits = 0;var result = mResult; mResult = 0;return (Int32)result; }}

publicstaticshortSwapInt16(short v){return (short)(((v & 0xFF) << 8) | ((v >> 8) & 0xFF));}publicstaticushortSwapUInt16(ushort v){return (ushort)((uint)((v & 0xFF) << 8) | ((uint)(v >> 8) & 0xFFu));}publicstaticintSwapInt32(int v){return ((SwapInt16((short)v) & 0xFFFF) << 16) | (SwapInt16((short)(v >> 16)) & 0xFFFF);}publicstaticuintSwapUInt32(uint v){return (uint)((SwapUInt16((ushort)v) & 0xFFFF) << 16) | (SwapUInt16((ushort)(v >> 16)) & 0xFFFFu);}

通过以上函数就可以进行对应存储字序的转换了，在读写流的时候加入这个函数调用即可。

publicvirtualvoidWriteUInt16(System.IO.Stream stream, ushortvalue){value = BitHelper.SwapUInt16(value);var data = BitConverter.GetBytes(value); stream.Write(data, 0, data.Length);}publicvirtualushortReadUInt16(System.IO.Stream stream){var buffer = GetIntBuffer(); stream.Read(buffer, 0, 2);var result = BitConverter.ToUInt16(buffer, 0);return BitHelper.SwapUInt16(result);}publicvirtualvoidWriteInt16(System.IO.Stream stream, shortvalue){value = BitHelper.SwapInt16(value);var data = BitConverter.GetBytes(value); stream.Write(data, 0, data.Length);}publicvirtualshortReadInt16(System.IO.Stream stream){var buffer = GetIntBuffer(); stream.Read(buffer, 0, 2);var result = BitConverter.ToInt16(buffer, 0);return BitHelper.SwapInt16(result);}

publicvirtualvoidWriteString(System.IO.Stream stream, stringvalue, Encoding encoding = null){if (encoding == null) encoding = Encoding.UTF8;if (string.IsNullOrEmpty(value)) { WriteUInt16(stream, 0);return; }byte[] buffer = System.Buffers.ArrayPool<byte>.Shared.Rent(value.Length * 6);var count = encoding.GetBytes(value, 0, value.Length, buffer, 0); WriteUInt16(stream, (ushort)count); stream.Write(buffer, 0, count); System.Buffers.ArrayPool<byte>.Shared.Return(buffer);}publicvirtualstringReadString(System.IO.Stream stream, Encoding encoding = null){if (encoding == null) encoding = Encoding.UTF8; UInt16 len = ReadUInt16(stream);if (len == 0)returnstring.Empty;byte[] buffer = System.Buffers.ArrayPool<byte>.Shared.Rent(len); stream.Read(buffer, 0, len);string result = encoding.GetString(buffer, 0, len); System.Buffers.ArrayPool<byte>.Shared.Return(buffer);return result;}

publicvoidWriteBinary(System.IO.Stream stream, byte[] data){ WriteUInt16(stream, (ushort)data.Length); stream.Write(data, 0, data.Length);}publicbyte[] ReadBinary(System.IO.Stream stream){var len = ReadUInt16(stream);byte[] result = newbyte[len]; stream.Read(result, 0, len);return result;}

以上是MQTT的协议类型描述和对应的实现操作，接下来就可以实现具体的协议了。

拆分消息包
在编写代码前先了解一下MQTT的消息包是如何拆分的，以下是协议定义的格式

协议把第一个字节拆分成两部分，第一部分的4比特用于描述消息相关的控制类型，不同消息包有不同的情况具体查看协议文档；第二部分是高4位比特用于描述消息类型。从第二个字节开始是一个变长整型用于描述消息体的剩下的数据长度。
其实从实现性能上来说我不认同这种协议设计，可变长度虽然可以节省2个节字的带宽(5.0协议是2019年定的，个人感觉现有的网络资源并不缺这点流量).....这种设计导致消息写入网络流存在一次内存拷贝，然而后面的扩展属性也是，这样一来导致一个消息的写入流就存在多次拷贝影响性能。下面来简单讲述这情况：

由于可变长度是根据消息内容来确定，导致无法在网络缓冲内存中分配一个固定的空间，因此只能在新的内存中写入内容得到长度后再把长度写入的网络内存块，然后再把对应的内容再复制到网络内存块中。如果长度是固定的如4字节，就可以在写入消息头后马上分配一个4字节的空间后直接写入具体的消息内容，当内容写入后得到的长再反填充之前分配的长度空间即可；这样的好处是消息内容可以直接写入到网络内存块中，无须用其他内存写入后再拷贝。

实现消息类型
既然有了协议的基础规则，那就可以针对它来实现一个消息抽像类来表达数据消息了。

publicabstract classMQTTMessage {publicMQTTMessage() { }publicabstract MQTTMessageType Type { get; }publicbyte Bit1 { get; set; }publicbyte Bit2 { get; set; }publicbyte Bit3 { get; set; }publicbyte Bit4 { get; set; }internalvoidRead(MQTTParse parse, System.IO.Stream stream, ISession session) { OnRead(parse, stream, session); }protectedvirtualvoidOnRead(MQTTParse parse, Stream stream, ISession session) { }internalvoidWrite(MQTTParse parse, System.IO.Stream stream, ISession session) { OnWrite(parse, stream, session); }protectedvirtualvoidOnWrite(MQTTParse parse, Stream stream, ISession session) { }}

在设计过程中预留OnRead和OnWrite方法给具体消息实现对应的网络数据读写；有了这个抽象的消息结构就可以对协议进行读写拆分了。

publicoverride MQTTMessage Read(Stream stream, ISession session){ IServer server = session?.Server;if (stream.Length > 0) {if (mType == null) { mHeader = (byte)stream.ReadByte();if (mHeader < 0) {thrownew BXException("parse mqtt message error:fixed header byte value cannot be less than zero!"); } mType = (MQTTMessageType)((mHeader & 0b1111_0000) >> 4);if (server != null && server.EnableLog(EventArgs.LogType.Debug)) server.Log(EventArgs.LogType.Debug, session, "parse mqtt header souccess"); }if (mLength == null) { mLength = mInt7BitHandler.Read(stream);if (server != null && server.EnableLog(EventArgs.LogType.Debug)) server.Log(EventArgs.LogType.Debug, session, $"parse mqtt size {mLength}"); }if (mLength != null && stream.Length >= mLength.Value) { Stream protocolStream = GetProtocolStream(); CopyStream(stream, protocolStream, mLength.Value); MQTTMessage msg = CreateMessage(mType.Value, session); msg.Bit1 = (byte)(BIT_1 & mHeader); msg.Bit2 = (byte)(BIT_2 & mHeader); msg.Bit3 = (byte)(BIT_3 & mHeader); msg.Bit4 = (byte)(BIT_4 & mHeader); msg.Read(this, protocolStream, session);if (server != null && server.EnableLog(EventArgs.LogType.Debug)) server.Log(EventArgs.LogType.Debug, session, $"parse mqtt type {msg} success"); mLength = null; mType = null;return msg; } }returnnull;}publicoverridevoidWrite(MQTTMessage msg, Stream stream, ISession session){ IServer server = session?.Server;if (server != null && server.EnableLog(EventArgs.LogType.Debug)) server.Log(EventArgs.LogType.Debug, session, $"write mqtt message {msg}");var protocolStream = GetProtocolStream();int header = 0; header |= ((int)msg.Type << 4); header |= msg.Bit1; header |= msg.Bit2 << 1; header |= msg.Bit3 << 2; header |= msg.Bit4 << 3; stream.WriteByte((byte)header); msg.Write(this, protocolStream, session);if (server != null && server.EnableLog(EventArgs.LogType.Debug)) server.Log(EventArgs.LogType.Debug, session, $"write mqtt message body size {protocolStream.Length}"); mInt7BitHandler.Write(stream, (int)protocolStream.Length); protocolStream.Position = 0; protocolStream.CopyTo(stream);if (server != null && server.EnableLog(EventArgs.LogType.Debug)) server.Log(EventArgs.LogType.Debug, session, $"write mqtt message success");}

通过以上的Read和Write方法就可以把网络数据和程序对象进行一个转换了。
到这里MQTT最上层的消息包解释处理就完成了，剩下不同消息的实现留给后面的章节再详细讲解。

BeetleX

开源跨平台通讯框架(支持TLS)

提供HTTP,Websocket,MQTT,Redis,RPC和服务网关开源组件

个人微信:henryfan128 QQ:28304340

关注公众号

https://github.com/beetlex-io/
http://beetlex-io.com