什么是MQTT协议并附GO语言实现示例

作者: memory

时间: 2024-10-25

MQTT（Message Queuing Telemetry Transport）是一种轻量级的消息传输协议，专为物联网（IoT）设备设计。它基于发布/订阅（Publish/Subscribe）模式，具有低带宽、低功耗、低开销的特点，非常适合在网络带宽有限或设备资源受限的环境中使用。

MQTT的主要特点：

发布/订阅模式：消息发布者（Publisher）将消息发布到主题（Topic），订阅者（Subscriber）订阅感兴趣的主题以接收消息。
QoS（Quality of Service）：MQTT支持三种服务质量级别：
- QoS 0：最多一次（At most once），消息可能丢失。
- QoS 1：至少一次（At least once），消息保证送达，但可能重复。
- QoS 2：恰好一次（Exactly once），消息保证送达且不重复。
遗嘱消息（Last Will and Testament, LWT）：客户端可以设置遗嘱消息，当客户端异常断开连接时，服务器会发送该消息。
保留消息（Retained Messages）：服务器可以保留最后一条发布到某个主题的消息，新订阅者订阅该主题时会立即收到这条消息。

GO语言实现MQTT示例

下面是一个使用Go语言实现MQTT协议的简单示例，包括发送和接收消息的功能。

1. 安装MQTT库

首先，你需要安装一个MQTT客户端库。在Go语言中，常用的MQTT库是paho.mqtt.golang。

go get -u github.com/eclipse/paho.mqtt.golang

2. 发送消息示例

package main

import (
    "fmt"
    "log"
    "os"
    "time"

    mqtt "github.com/eclipse/paho.mqtt.golang"
)

func main() {
    // 创建MQTT客户端选项
    opts := mqtt.NewClientOptions()
    opts.AddBroker("tcp://broker.hivemq.com:1883") // MQTT Broker地址
    opts.SetClientID("go_mqtt_client")             // 客户端ID

    // 设置连接丢失时的回调函数
    opts.OnConnectionLost = func(client mqtt.Client, err error) {
        log.Println("连接丢失:", err)
    }

    // 创建MQTT客户端
    client := mqtt.NewClient(opts)
    if token := client.Connect(); token.Wait() && token.Error() != nil {
        log.Fatalf("连接失败: %v", token.Error())
    }

    // 发布消息
    topic := "test/topic"
    message := "Hello, MQTT!"
    token := client.Publish(topic, 0, false, message)
    token.Wait()

    fmt.Printf("消息已发布到主题 %s: %s\n", topic, message)

    // 断开连接
    client.Disconnect(250)
}

3. 接收消息示例

package main

import (
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"

    mqtt "github.com/eclipse/paho.mqtt.golang"
)

func main() {
    // 创建MQTT客户端选项
    opts := mqtt.NewClientOptions()
    opts.AddBroker("tcp://broker.hivemq.com:1883") // MQTT Broker地址
    opts.SetClientID("go_mqtt_subscriber")         // 客户端ID

    // 设置连接丢失时的回调函数
    opts.OnConnectionLost = func(client mqtt.Client, err error) {
        log.Println("连接丢失:", err)
    }

    // 创建MQTT客户端
    client := mqtt.NewClient(opts)
    if token := client.Connect(); token.Wait() && token.Error() != nil {
        log.Fatalf("连接失败: %v", token.Error())
    }

    // 订阅主题
    topic := "test/topic"
    if token := client.Subscribe(topic, 0, func(client mqtt.Client, msg mqtt.Message) {
        fmt.Printf("收到消息: 主题: %s, 消息: %s\n", msg.Topic(), string(msg.Payload()))
    }); token.Wait() && token.Error() != nil {
        log.Fatalf("订阅失败: %v", token.Error())
    }

    fmt.Printf("已订阅主题 %s\n", topic)

    // 等待中断信号以优雅地关闭连接
    sig := make(chan os.Signal, 1)
    signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
    <-sig

    // 断开连接
    client.Disconnect(250)
}

代码注解

MQTT客户端选项：
- AddBroker：设置MQTT Broker的地址。
- SetClientID：设置客户端ID，用于标识客户端。
- OnConnectionLost：设置连接丢失时的回调函数。
连接MQTT Broker：
- client.Connect()：连接到MQTT Broker，返回一个token对象，通过token.Wait()和token.Error()检查连接是否成功。
发布消息：
- client.Publish(topic, qos, retained, message)：发布消息到指定主题，qos为服务质量级别，retained表示是否保留消息。
订阅主题：
- client.Subscribe(topic, qos, callback)：订阅指定主题，callback为收到消息时的回调函数。
断开连接：
- client.Disconnect(quiesce)：断开连接，quiesce为等待时间（毫秒）。