目前编程网上已经有很多关于Golang的文章了，自己在初次阅读这些文章中，也见识到了很多学习思路；那么本文《MQTT paho - 发布消息时未出现错误》，也希望能帮助到大家，如果阅读完后真的对你学习Golang有帮助，欢迎动动手指，评论留言并分享~
问题内容我正在使用 paho.mqtt.golang 库连接到代理并发布消息。
它工作正常，只是发布失败时没有出现错误。
我正在做的测试如下：
- 我启动经纪人
- 我运行代码来连接到代理。连接后，代码等待输入以继续发布
- 我杀了经纪人
- 我按 enter 键继续发布消息
我预计发布函数 if token.error() != nil {...} 返回的令牌会出现错误，但我没有得到任何错误。
这是发布函数的代码：
func (handle handler) pub(ctx context.context, topic, payload string, qos int, retained bool) error {
token := handle.client.publish(topic, byte(qos), retained, payload)
go func(ctx context.context) {
log := logger.newlogfromctx(ctx)
log.debug("waiting for transaction to complete...")
_ = token.done()
log.debug("transaction done!", zap.any("token.error()", token.error()))
if token.error() != nil {
log.error("failed to publish mqtt message", zap.error(token.error()))
}
}(ctx)
log := logger.newlogfromctx(ctx)
log.debug("msg sent !")
return nil
}这是日志：
thu 27 may 17:40:25 cest info logger/logging.go:32 logger initialized in development mode
[debug][client] connect()
[debug][store] memorystore initialized
[debug][client] about to write new connect msg
[debug][client] socket connected to broker
[debug][client] using mqtt 3.1.1 protocol
[debug][net] connect started
[debug][net] received connack
[debug][client] startcommsworkers called
[debug][client] client is connected/reconnected
[debug][net] incoming started
[debug][net] startincomingcomms started
[debug][net] outgoing started
[debug][net] startcomms started
[debug][client] startcommsworkers done
[warn][store] memorystore wiped
[debug][client] exit startclient
thu 27 may 17:40:25 cest info mqtt_/client.go:68 successfully connected to mqtt broker {"url": "tcp://127.0.0.1:1883", "in": "41.843622ms"}
press enter to publish...此时我已连接到代理，代码正在等待输入；我杀了经纪人：
[error][client] connect comms goroutine - error triggered eof
[debug][client] internalconnlost called
[debug][client] stopcommsworkers called
[debug][router] matchanddispatch exiting
[debug][pinger] keepalive stopped
[debug][client] startcommsworkers output redirector finished
[debug][net] outgoing waiting for an outbound message
[debug][net] outgoing waiting for an outbound message
[debug][net] outgoing comms stopping
[debug][net] startcomms closing outerror
[debug][client] incoming comms goroutine done
[debug][client] stopcommsworkers waiting for workers
[debug][client] stopcommsworkers waiting for comms
[debug][client] stopcommsworkers done
[debug][client] internalconnlost waiting on workers
[debug][client] internalconnlost workers stopped
[debug][client] internalconnlost complete
[debug]connection lost: eof
[debug][client] enter reconnect
[debug][client] about to write new connect msg
[debug][client] socket connected to broker
[debug][client] using mqtt 3.1.1 protocol
[debug][net] connect started
[error][net] connect got error eof
[error][client] connecting to tcp://127.0.0.1:1883 connack was not conn_accepted, but rather connection error
[debug][client] reconnect failed, sleeping for 1 seconds: network error : eof连接确实丢失了。我按 enter 键继续发布：
[debug][client] enter publish
[debug][client] storing publish message (reconnecting), topic: just/for/test
thu 27 may 17:40:42 cest debug mqtt_/client.go:84 msg sent !
thu 27 may 17:40:42 cest debug mqtt_/client.go:76 waiting for transaction to complete...
thu 27 may 17:40:42 cest debug mqtt_/client.go:78 transaction done! {"token.error()": null}token.error() 中没有任何内容。如何检查发布是否成功？
如果您需要更多详细信息，这是我的完整代码。
连接并发布到代理：
type handler struct {
client mqtt.client
conf config.configuration
}
func initmqtt() {
mqtt.debug = lg.new(os.stdout, "[debug]", 0)
mqtt.warn = lg.new(os.stdout, "[warn]", 0)
mqtt.critical = lg.new(os.stdout, "[crit]", 0)
mqtt.error = lg.new(os.stdout, "[error]", 0)
}
func newclient(ctx context.context, conf config.configuration) (handler, error) {
start := time.now()
log := logger.newlogfromctx(ctx)
brokerurl := fmt.sprintf("tcp://%s:%s", conf.gw_mqtt_broker_host_ip, conf.gw_mqtt_broker_port)
hostname, _ := os.hostname()
clientid := hostname + strconv.itoa(time.now().second())
connopts := mqtt.newclientoptions()
connopts.addbroker(brokerurl)
connopts.setclientid(clientid)
connopts.setcleansession(true)
handler := handler{conf: conf}
handler.client = mqtt.newclient(connopts)
if token := handler.client.connect(); token.wait() && token.error() != nil {
log.error("failed to connect to mqtt broker", zap.error(token.error()))
return handler{}, token.error()
}
log.info("successfully connected to mqtt broker", zap.string("url", brokerurl), zap.duration("in", time.since(start)))
return handler, nil
}
func (handle handler) pub(ctx context.context, topic, payload string, qos int, retained bool) error {
token := handle.client.publish(topic, byte(qos), retained, payload)
go func(ctx context.context) {
log := logger.newlogfromctx(ctx)
log.debug("waiting for transaction to complete...")
_ = token.done()
log.debug("transaction done!", zap.any("token.error", token.error()))
if token.error() != nil {
log.error("failed to publish mqtt message", zap.error(token.error()))
}
}(ctx)
log := logger.newlogfromctx(ctx)
log.debug("msg sent !")
return nil
}这是主要内容：
func main() {
conf := config.GetConfig()
err := logger.SetupLogging(conf.IS_DEV_ENV)
if err != nil {
panic(err)
}
ctx := context.Background()
log := logger.NewLogFromCtx(ctx)
mqtt.InitMQTT()
mqttClient, _ := mqtt.NewClient(ctx, conf)
reader := bufio.NewReader(os.Stdin)
fmt.Print("press enter to publish...")
text, _ := reader.ReadString('\n')
mqttClient.Pub(ctx, "just/for/test", "test", 2, false)
}正确答案
来自 the docs：
// wait will wait indefinitely for the token to complete, ie the publish
// to be sent and confirmed receipt from the broker.
wait() bool// Done is provided for use in select statements. Simple use cases may
// use Wait or WaitTimeout.
Done() <-chan struct{}所以 _ = token.done() 实际上什么也没做；最简单的等待是使用 token.wait()。如果您想使用 token.done() 您需要等待返回的通道；例如<- token.done()。 done() 存在的原因是为了让您在等待多个事件时变得更简单（例如，等待 context 或 mqtt 操作完成）。
今天关于《MQTT paho - 发布消息时未出现错误》的内容介绍就到此结束，如果有什么疑问或者建议，可以在编程网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！
