1.MQTT协议:
MQTT(Message Queuing Telemetry Transport,消息队列遥测传输协议),是一种基于TCP/IP实现发布/订阅模式的应用层协议,其主要特点有:
(1)基于发布/订阅模式,应用程序解耦;
(2)基于TCP/IP建立网络连接;
(3)协议交换最小化,降低网络流量;
2.基于MQTT协议应用:
(1)简单的发布订阅应用:
(2)消息系统推送应用:
(3)阿里云物联网应用:
由于知识能力受限,无法一一列举基于MQTT协议的各种应用,下面就以上述消息推送系统作为例子,讲讲基于MQTT协议的消息推送系统具体开发,不考虑太复杂的业务逻辑,仅以最简洁的方式,阐述整个流程,针对一款应用来说,以一百万用户在线作为设计目标,基于golang/c/c++/javascript开发.
3.MQTT控制报文组成结构:
3.1FixedHeader(固定头部)结构:
针对MQTT固定头部定义相关数据结构,并定义相关编解码方法,如下:
type FixedHeader struct { MessageType byte Dup bool Qos byte Retain bool RemainingLength int}func boolToByte(b bool) byte { switch b { case true: return 1 default: return 0 }}//编码固定头部func (fh *FixedHeader) pack() bytes.Buffer { var header bytes.Buffer header.WriteByte(fh.MessageType<<4 | boolToByte(fh.Dup)<<3 | fh.Qos<<1 | boolToByte(fh.Retain)) header.Write(encodeLength(fh.RemainingLength)) return header}//解码固定头部func (fh *FixedHeader) unpack(typeAndFlags byte, r io.Reader) { fh.MessageType = typeAndFlags >> 4 fh.Dup = (typeAndFlags>>3)&0x01 > 0 fh.Qos = (typeAndFlags >> 1) & 0x03 fh.Retain = typeAndFlags&0x01 > 0 fh.RemainingLength = decodeLength(r)}//编码剩余长度func encodeLength(length int) []byte { var encLength []byte for { digit := byte(length % 128) length /= 128 if length > 0 { digit |= 0x80 } encLength = append(encLength, digit) if length == 0 { break } } return encLength}//解码剩余长度func decodeLength(r io.Reader) int { var rLength uint32 var multiplier uint32 b := make([]byte, 1) for { io.ReadFull(r, b) digit := b[0] rLength |= uint32(digit&127) << multiplier if (digit & 128) == 0 { break } multiplier += 7 } return int(rLength)}
3.2VariableHeader(可变头部)结构:
可变头部结构根据请求报文的不同而不同,下面以CONNECT报文为例讲述,CONNECT报文可变头部结构:协议名称:
协议级别:
连接标志:
保持连接:
3.3Payload(有效负荷):
有效负荷根据请求报文的不同而不同,下面以CONNECT报文为例讲述,CONNECT报文可变头部结构,CONNECT报文的有效载荷包含一个或多个以长度为前缀的字段,由可变报头中的标志决定是否包含这些字段,字段必须按这个顺序出现:客户端标识符,遗嘱主题,遗嘱消息,用户名,密码.3.4连接报文编解码:
综上,针对MQTT连接报文定义相关数据结构,并定义相关编解码方法,如下:type ConnectPacket struct { FixedHeader ProtocolName string ProtocolVersion byte CleanSession bool WillFlag bool WillQos byte WillRetain bool UsernameFlag bool PasswordFlag bool ReservedBit byte KeepaliveTimer uint16 ClientIdentifier string WillTopic string WillMessage []byte Username string Password []byte}//连接报文编码func (c *ConnectPacket) Write(w io.Writer) error { var body bytes.Buffer var err error body.Write(encodeString(c.ProtocolName)) body.WriteByte(c.ProtocolVersion) body.WriteByte(boolToByte(c.CleanSession)<<1 | boolToByte(c.WillFlag)<<2 | c.WillQos<<3 | boolToByte(c.WillRetain)<<5 | boolToByte(c.PasswordFlag)<<6 | boolToByte(c.UsernameFlag)<<7) body.Write(encodeUint16(c.KeepaliveTimer)) body.Write(encodeString(c.ClientIdentifier)) if c.WillFlag { body.Write(encodeString(c.WillTopic)) body.Write(encodeBytes(c.WillMessage)) } if c.UsernameFlag { body.Write(encodeString(c.Username)) } if c.PasswordFlag { body.Write(encodeBytes(c.Password)) } c.FixedHeader.RemainingLength = body.Len() packet := c.FixedHeader.pack() packet.Write(body.Bytes()) _, err = packet.WriteTo(w) return err}//连接报文解码func (c *ConnectPacket) Unpack(b io.Reader) { c.ProtocolName = decodeString(b) c.ProtocolVersion = decodeByte(b) options := decodeByte(b) c.ReservedBit = 1 & options c.CleanSession = 1&(options>>1) > 0 c.WillFlag = 1&(options>>2) > 0 c.WillQos = 3 & (options >> 3) c.WillRetain = 1&(options>>5) > 0 c.PasswordFlag = 1&(options>>6) > 0 c.UsernameFlag = 1&(options>>7) > 0 c.KeepaliveTimer = decodeUint16(b) c.ClientIdentifier = decodeString(b) if c.WillFlag { c.WillTopic = decodeString(b) c.WillMessage = decodeBytes(b) } if c.UsernameFlag { c.Username = decodeString(b) } if c.PasswordFlag { c.Password = decodeBytes(b) }}
出于篇幅考虑,上述使用到的具体一些函数,如decodeString,decodeByte,encodeString等,就不一一列举出来了,如有错误,恳请指出,转载也请注明出处!!!
未完待续...参考文字: