基於MQTT協議談談物聯網開發

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等,就不一一列舉出來了,如有錯誤,懇請指出,轉載也請注明出處!!!

未完待續...
參考文字:MQTT協議中文版