我正在尝试使用paho pkg来通过golang构建MQTT子客户端,当代理断开连接时,我的客户端就会出现问题,我一开始以为会出现丢失的消息,但这种情况并没有发生,启动代理后,MQTT子客户端却无法获得由MQTT PUB客户端发送的消息。
为什么会发生这种事,我怎么才能解决呢?
代码:
package main
import (
"fmt"
"os"
mqtt "github.com/eclipse/paho.mqtt.golang"
)
var (
broker = "tcp://localhost:1883"
f mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
fmt.Printf("TOPIC: %s
", msg.Topic())
fmt.Printf("MSG: %s
", msg.Payload())
}
)
func main() {
//create a ClientOptions
opts := mqtt.NewClientOptions().AddBroker(broker)
opts.SetClientID("group-one")
opts.SetDefaultPublishHandler(f)
//create and start a client using the above ClientOptions
c := mqtt.NewClient(opts)
if token := c.Connect(); token.Wait() && token.Error() != nil {
panic(token.Error())
}
if token := c.Subscribe("test", 0, nil); token.Wait() && token.Error() != nil {
fmt.Println(token.Error())
os.Exit(1)
}
for {
}
}
Assign a custom OnConnectionLostHandler
to catch connection lost event, so you can perform additional action whenever the client loses connection. If you set the AutoReconnect
option to true
(which is the default behavior), the client will automatically reconnects to the broker after connection lost. Please note that after connection lost, your subscription state/info may not being saved by the broker, so you won't be able to receive any message. To deal with this issue, move topic subscription to OnConnect
handler. Below is an example implementation:
package main
import (
"fmt"
"os"
"time"
mqtt "github.com/eclipse/paho.mqtt.golang"
)
func messageHandler(c mqtt.Client, msg mqtt.Message) {
fmt.Printf("TOPIC: %s
", msg.Topic())
fmt.Printf("MSG: %s
", msg.Payload())
}
func connLostHandler(c mqtt.Client, err error) {
fmt.Printf("Connection lost, reason: %v
", err)
//Perform additional action...
}
func main() {
//create a ClientOptions
opts := mqtt.NewClientOptions().
AddBroker("tcp://localhost:1883").
SetClientID("group-one").
SetDefaultPublishHandler(messageHandler).
SetConnectionLostHandler(connLostHandler)
//set OnConnect handler as anonymous function
//after connected, subscribe to topic
opts.OnConnect = func(c mqtt.Client) {
fmt.Printf("Client connected, subscribing to: test/topic
")
//Subscribe here, otherwise after connection lost,
//you may not receive any message
if token := c.Subscribe("test/topic", 0, nil); token.Wait() && token.Error() != nil {
fmt.Println(token.Error())
os.Exit(1)
}
}
//create and start a client using the above ClientOptions
c := mqtt.NewClient(opts)
if token := c.Connect(); token.Wait() && token.Error() != nil {
panic(token.Error())
}
for {
//Lazy...
time.Sleep(500 * time.Millisecond)
}
}