基于MQTT协议谈谈物联网开发

1.MQTT协议:

MQTT(Message Queuing Telemetry Transport,消息队列遥测传输协议),是一种基于TCP/IP实现发布/订阅模式的应用层协议,其主要特点有:

(1)基于发布/订阅模式,应用程序解耦;

(2)基于TCP/IP建立网络连接;

(3)协议交换最小化,降低网络流量;

2.基于MQTT协议应用:

(1)简单的发布订阅应用:

MQTT发布订阅应用

(2)消息系统推送应用:

MQTT消息推送应用

(3)阿里云物联网应用:

阿里云物联网应用

由于知识能力受限,无法一一列举基于MQTT协议的各种应用,下面就以上述消息推送系统作为例子,讲讲基于MQTT协议的消息推送系统具体开发,不考虑太复杂的业务逻辑,仅以最简洁的方式,阐述整个流程,针对一款应用来说,以一百万用户在线作为设计目标,基于golang/c/c++/javascript开发.

3.MQTT控制报文组成结构:

MQTT控制报文

3.1FixedHeader(固定头部)结构:

MQTT固定头部

针对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
    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) 

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) > 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等,就不一一列举出来了,如有错误,恳请指出!!!

未完待续…
参考文字:MQTT协议中文版