MQTT Go 客户端库

import (
    "fmt"
    "log"
    "os"
    "time"
    "github.com/eclipse/paho.mqtt.golang"
)
var f mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
    fmt.Printf("TOPIC: %s\n", msg.Topic())
    fmt.Printf("MSG: %s\n", msg.Payload())
}
func main() {
    mqtt.ERROR = log.New(os.Stdout, "", 0)
    opts := mqtt.NewClientOptions().AddBroker("tcp://broker.emqx.io:1883").SetClientID("emqx_test_client")
    opts.SetKeepAlive(60 * time.Second)
    // 设置消息回调处理函数
    opts.SetDefaultPublishHandler(f)
    opts.SetPingTimeout(1 * time.Second)
    c := mqtt.NewClient(opts)
    if token := c.Connect(); token.Wait() && token.Error() != nil {
        panic(token.Error())
    }
    // 订阅主题
    if token := c.Subscribe("testtopic/#", 0, nil); token.Wait() && token.Error() != nil {
    }
    // 发布消息
    token := c.Publish("testtopic/1", 0, false, "Hello World")
    token.Wait()
    time.Sleep(6 * time.Second)
    // 取消订阅
    if token := c.Unsubscribe("testtopic/#"); token.Wait() && token.Error() != nil {
        fmt.Println(token.Error())
        os.Exit(1)
    }
  // 断开连接
    c.Disconnect(250)
}

Paho Golang MQTT 5.0 支持