MQTT paho - 发布消息时未出现错误

目前编程网上已经有很多关于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 - 发布消息时未出现错误》的内容介绍就到此结束，如果有什么疑问或者建议，可以在编程网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！