1. 加載mqtt包

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

2. 初始化MQTT服務(wù)

var client mqtt.Client

// 初始化MQTT服務(wù)
func NewClient() {
   if client == nil {
       opts := mqtt.NewClientOptions()
       opts.AddBroker("tcp://broker.emqx.io:1883") // 這個(gè)中轉(zhuǎn)服務(wù)器不需要任何賬號(hào)密碼
       opts.SetClientID("go_mqtt_client1")
       // opts.SetUsername("")
       // opts.SetPassword("")
       opts.OnConnect = func(c mqtt.Client) {
           fmt.Println("MQTT鏈接成功！")
       }
       opts.OnConnectionLost = func(c mqtt.Client, err error) {
           fmt.Println("MQTT斷開鏈接！", err.Error())
           fmt.Println("嘗試重新鏈接！")
           NewClient()
       }
       client = mqtt.NewClient(opts)
   }

   if token := client.Connect(); token.Wait() && token.Error() != nil {
       fmt.Println(token.Error())
   }

   // 訂閱事件
   for _, subscribe := range subscribes {
       subscribe()
   }
}

3. 發(fā)布消息與訂閱消息調(diào)用方法

// 發(fā)布消息 ClientSend("topic/publish/example", 2, false, `{"code":0, "msg":"hello world!"}`)
func ClientSend(topic string, qos byte, retained bool, payload interface{}) error {
    if token := client.Publish(topic, qos, retained, payload); token.Wait() && token.Error() != nil {
        fmt.Println("消息發(fā)布失敗！", token.Error())
        return token.Error()
    }
    return nil
}

// 訂閱消息
func ClientSubscribe(topic string, qos byte, callback mqtt.MessageHandler, err func(error)) {
    if token := client.Subscribe(topic, qos, func(c mqtt.Client, msg mqtt.Message) {
        callback(c, msg)
    }); token.Wait() && token.Error() != nil {
        err(token.Error())
    }
}

4. 發(fā)布消息

err := ClientSend("topic/publish/example", 2, false, `{"code":0, "msg":"hello world!"}`)
fmt.Println(err)

5. 訂閱消息

// 訂閱消息
var subscribes = []func(){
    // 直接寫方法
    func() {
        ClientSubscribe("topic/subscribe/example", 1, func(c mqtt.Client, msg mqtt.Message) {
            fmt.Println("subscribe Msg：", string(msg.Payload()))
        }, func(err error) {
            fmt.Println(err.Error())
        })
    },
    // 調(diào)用
    subscribeExample2,
}

func subscribeExample2() {
    ClientSubscribe("topic/subscribe/example2", 1, func(c mqtt.Client, msg mqtt.Message) {
        fmt.Println("subscribe Msg2：", string(msg.Payload()))
    }, func(err error) {
        fmt.Println(err.Error())
    })
}