未收到其他客户端发送的消息的MQTT客户端

欢迎各位小伙伴来到golang学习网，相聚于此都是缘哈哈哈！今天我给大家带来《未收到其他客户端发送的消息的MQTT客户端》，这篇文章主要讲到等等知识，如果你对Golang相关的知识非常感兴趣或者正在自学，都可以关注我，我会持续更新相关文章！当然，有什么建议也欢迎在评论留言提出！一起学习！

我遇到一个问题，在几次断开连接和重新连接后，我的 mqtt 客户端（比如说 a）停止接收来自另一个发布客户端（b）的消息。

如果我使用 mosquitto_sub 手动订阅 b 正在发布的主题，我可以看到所有消息都按预期发布。如果我手动发布（mosquitto_pub）到 a 订阅的主题，a 也会收到这些消息，因此订阅似乎有效。仅当 b 正在发布到主题时，a 才不会接收消息。

两个客户端都连接到 mosquitto 代理（版本 1.6.12）。一切都在 raspberrypi cm3 上运行。客户端使用 paho mqtt 库 (v1.4.2) 用 go 编写，并作为 systemd 服务启动。客户端使用以下选项进行初始化：

opts.setcleansession(false)
opts.setkeepalive(10 * time.second)
opts.setpingtimeout(1 * time.second)
opts.setautoreconnect(true)
opts.setconnecttimeout(15 * time.second)

此外，两个客户端都会获得一个唯一的 id。 我查看了 mosquitto 日志，注意到在他订阅之前，一些消息被发送到重新连接的客户端。

1678786214: sending publish to telemetry (d1, q1, r0, m722, 'example-topic-1/a', ... (109 bytes))
1678786214: sending publish to telemetry (d1, q1, r0, m723, 'example-topic-1/b', ... (118 bytes))
1678786214: sending publish to telemetry (d1, q1, r0, m904, 'example-topic-1/a', ... (109 bytes))
1678786214: sending publish to telemetry (d1, q1, r0, m1085, 'example-topic-1/a', ... (109 bytes))
1678786214: sending publish to telemetry (d1, q1, r0, m1086, 'example-topic-1/b', ... (118 bytes))
1678786214: sending publish to telemetry (d1, q1, r0, m1267, 'example-topic-1/a', ... (109 bytes))
...
...
1678786214: sending publish to telemetry (d0, q1, r0, m2293, 'example-topic-1/c'1, ... (119 bytes))
1678786214: received publish from telemetry (d0, q2, r1, m1, 'example-topic-2', ... (7 bytes))
1678786214: sending pubrec to telemetry (m1, rc0)
1678786214: received publish from telemetry (d0, q2, r1, m2, 'example-topic-1/a', ... (109 bytes))
1678786214: sending pubrec to telemetry (m2, rc0)
1678786214: received pubrel from telemetry (mid: 1)
1678786214: sending pubcomp to telemetry (m1)
1678786214: received pubrel from telemetry (mid: 2)
1678786214: sending pubcomp to telemetry (m2)
1678786214: received subscribe from telemetry
1678786214:         shutdown (qos 2)
1678786214: telemetry 2 shutdown
1678786214: sending suback to telemetry
1678786214: received subscribe from telemetry
1678786214:         example-topic-1/# (qos 1)
1678786214: telemetry 1 example-topic-1/#
1678786214: sending suback to telemetry

在 mqtt 库的文档中它指出

when qos1+ subscriptions have been created previously and you connect with cleansession set to false it is possible that the broker will deliver retained messages before subscribe can be called. to process these messages either configure a handler with addroute or set a defaultpublishhandler.

这是否意味着消息将丢失，或者它实际上以某种方式阻止了主题？

到目前为止，我尝试将 mosquitto 从最初的版本 1.6.10 更新到 1.6.12，但没有解决问题。重新启动 mosquitto.service 似乎可以解决问题，但并不是真正的解决方案。我的下一步是设置 defaultpublishhandler 来处理过早发送的消息。

预先感谢您提供的任何帮助。如果需要更多信息，请告诉我！

编辑：

我做了更多尝试，查看了 mosquitto 日志，可以稍微缩小问题范围，因为它似乎与消息的 qos 级别有关。

虽然客户端 a 没有接收客户端 b 正在发送的任何消息，但我尝试注册一个新客户端（使用 mosquitto_sub），我们称其为 c，无论订阅的 qos 级别如何，它都会接收来自客户端 b 的消息。当使用客户端 c 进行发布时，客户端 a 仅接收 qos 为 0 发送的消息。如果我指定 qos 为 1 或 2，则客户端 a 不会收到任何消息。由于客户端 b 发送 qos 为 1 的所有消息，因此代理似乎没有向客户端 a 发送任何 qos > 0 的消息，而是向其他客户端发送。

所有客户端的 ordermatters 设置为 false，代理的唯一配置是 max_queued_messages 0。 以下是日志的一些相关片段。

clientb 使用 qos 1 进行发布

1678874048: sending puback to clientb (m732, rc0)
1678874048: received publish from clientb (d0, q1, r0, m733, 'topic/b', ... (128 bytes))
1678874048: sending puback to clientb (m733, rc0)
1678874048: received publish from clientb (d0, q1, r0, m734, 'topic/c', ... (135 bytes))
1678874048: sending puback to clientb (m734, rc0)

注意没有消息发布到 clienta。

clientc 发布，qos 1

1678874088: no will message specified.
1678874088: sending connack to mosq-glzkfjx79cric0qok5 (0, 0)
1678874088: received publish from mosq-glzkfjx79cric0qok5 (d0, q1, r0, m1, 'topic/a', ... (130 bytes))
1678874088: sending puback to mosq-glzkfjx79cric0qok5 (m1, rc0)
1678874088: received disconnect from mosq-glzkfjx79cric0qok5
1678874088: client mosq-glzkfjx79cric0qok5 disconnected.

如果使用新客户端发布，则相同。

clientc 发布，qos 0

1678874043: no will message specified.
1678874043: sending connack to mosq-pba5gdo7lne7sqz5jd (0, 0)
1678874043: received publish from mosq-pba5gdo7lne7sqz5jd (d0, q0, r0, m0, 'topic/a', ... (130 bytes))
1678874043: sending publish to clienta (d0, q0, r0, m0, 'topic/a', ... (130 bytes))
1678874043: received disconnect from mosq-pba5gdo7lne7sqz5jd
1678874043: client mosq-pba5gdo7lne7sqz5jd disconnected.

使用新客户端进行发布，qos 为 0 并发布到 clienta。

clientc 以 qos 1 订阅

1678874687: no will message specified.
1678874687: sending connack to mosq-vkwiratvyxhwolncqh (0, 0)
1678874687: received subscribe from mosq-vkwiratvyxhwolncqh
1678874687:         topic/+ (qos 1)
1678874687: mosq-vkwiratvyxhwolncqh 1 topic/+
1678874687: sending suback to mosq-vkwiratvyxhwolncqh
...
1678874695: received publish from clientb (d0, q1, r0, m673, 'topic/a', ... (137 bytes))
1678874695: sending puback to clientb (m673, rc0)
1678874695: sending publish to mosq-vkwiratvyxhwolncqh (d0, q1, r0, m3, 'topic/a', ... (137 bytes))
1678874695: received publish from clientb (d0, q1, r0, m674, 'topic/b', ... (130 bytes))
1678874695: sending puback to clientb (m674, rc0)
1678874695: received puback from mosq-vkwiratvyxhwolncqh (mid: 3, rc:0)
1678874695: sending publish to mosq-vkwiratvyxhwolncqh (d0, q1, r0, m4, 'topic/b', ... (130 bytes))
1678874695: received puback from mosq-vkwiratvyxhwolncqh (mid: 4, rc:0)

使用 qos 1 的新客户端订阅也可以...

老实说，我现在有点没有想法，因为在我看来，这种行为很奇怪。 callbackhandlers 似乎不会在 clientas 端阻塞，因为 qos 0 的消息仍然得到处理。是否有任何设置或配置可能会干扰 qos 1 订阅？

编辑2：

以下日志基本上是重新启动服务之前和之后的日志，这导致了所描述的行为。由于不仅仅是这两个客户不断发送东西，我再次剪掉了一些部分，但是有关这两个客户的所有内容都应该在这里。从第一次连接和一些工作发布到重新启动和不再发送消息。

我注意到的另一件事是，关闭服务时，客户端永远不会调用 disconnect。它尝试取消订阅，有时甚至无法完成。这会导致这样的问题吗？

1678880076: new client connected from 127.0.0.1 as clienta (p2, c0, k10).
1678880076: no will message specified.
1678880076: sending connack to clienta (1, 0)
1678880076: sending publish to clienta (d1, q1, r0, m18734, 'topic1/matlab_version_expected', ... (152 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m18735, 'topic1/coreagent_ota_state', ... (112 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m19039, 'topic1/telemetry_data_sent', ... (109 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m19041, 'topic1/telemetry_queue_size', ... (119 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m19331, 'topic1/telemetry_data_sent', ... (109 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m19333, 'topic1/telemetry_queue_size', ... (119 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m19453, 'topic1/telemetry_data_sent', ... (109 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m19454, 'topic1/telemetry_queue_size', ... (119 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m19575, 'topic1/telemetry_data_sent', ... (109 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m19576, 'topic1/telemetry_queue_size', ... (119 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m20053, 'topic1/telemetry_data_sent', ... (109 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m20054, 'topic1/telemetry_queue_size', ... (119 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m20235, 'topic1/telemetry_data_sent', ... (109 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m20236, 'topic1/telemetry_queue_size', ... (119 bytes))
1678880076: sending publish to clienta (d1, q1, r0, m20349, 'topic1/ostree_sha_rollback', ... (172 bytes))
... (lots of publish withouth puback from clienta)
1678880076: sending publish to clienta (d1, q1, r0, m20540, 'topic1/telemetry_queue_size', ... (119 bytes))
1678880076: received publish from clienta (d0, q2, r1, m1, 'conn-status', ... (7 bytes))
1678880076: sending pubrec to clienta (m1, rc0)
1678880076: received publish from clienta (d0, q2, r1, m2, 'topic1/telemetry_data_sent', ... (109 bytes))
1678880076: sending pubrec to clienta (m2, rc0)
1678880076: received pubrel from clienta (mid: 1)
1678880076: sending pubcomp to clienta (m1)
1678880076: received pubrel from clienta (mid: 2)
1678880076: sending pubcomp to clienta (m2)
1678880076: sending publish to clienta (d0, q1, r0, m20720, 'topic1/telemetry_data_sent', ... (109 bytes))
1678880076: received publish from clienta (d0, q2, r1, m3, 'topic1/telemetry_queue_size', ... (119 bytes))
1678880076: sending pubrec to clienta (m3, rc0)
1678880076: received subscribe from clienta
1678880076:         shutdown (qos 2)
1678880076: clienta 2 shutdown
1678880076: sending suback to clienta
1678880076: received pubrel from clienta (mid: 3)
1678880076: sending pubcomp to clienta (m3)
1678880076: sending publish to clienta (d0, q1, r0, m20721, 'topic1/telemetry_queue_size', ... (119 bytes))
1678880076: received subscribe from clienta
1678880076:         telemetry/+ (qos 1)
1678880076: clienta 1 telemetry/+
1678880076: sending suback to clienta
1678880076: sending publish to clienta (d0, q1, r1, m20722, 'telemetry/geolocation', ... (51 bytes))
1678880076: received puback from clienta (mid: 20722, rc:0)
1678880076: received subscribe from clienta
1678880076:         telemetry-batch (qos 1)
1678880076: clienta 1 telemetry-batch
1678880076: sending suback to clienta
1678880076: received subscribe from clienta
1678880076:         topic1/# (qos 1)
1678880076: clienta 1 topic1/#
1678880076: sending suback to clienta
1678880076: sending publish to clienta (d0, q1, r1, m20723, 'topic1/telemetry_data_sent', ... (109 bytes))
1678880076: sending publish to clienta (d0, q1, r1, m20724, 'topic1/telemetry_queue_size', ... (119 bytes))
1678880076: received puback from clienta (mid: 20723, rc:0)
1678880076: received puback from clienta (mid: 20724, rc:0)
1678880076: received publish from clienta (d0, q2, r1, m8, 'conn-status', ... (6 bytes))
1678880076: sending pubrec to clienta (m8, rc0)
1678880076: received pubrel from clienta (mid: 8)
1678880076: sending pubcomp to clienta (m8)
...
1678880083: new connection from 127.0.0.1 on port 1883.
1678880083: new client connected from 127.0.0.1 as clientb (p2, c0, k10).
1678880083: no will message specified.
1678880083: sending connack to clientb (1, 0)
1678880085: received publish from clienta (d0, q2, r1, m9, 'session', ... (36 bytes))
1678880085: sending pubrec to clienta (m9, rc0)
1678880085: received pubrel from clienta (mid: 9)
1678880085: sending pubcomp to clienta (m9)
...
1678880086: received publish from clientb (d0, q1, r0, m3, 'topic1/net_response', ... (118 bytes))
1678880086: sending puback to clientb (m3, rc0)
1678880086: sending publish to clienta (d0, q1, r0, m20725, 'topic1/net_response', ... (118 bytes))
1678880086: received puback from clienta (mid: 20725, rc:0)
1678880086: received publish from clientb (d0, q1, r0, m4, 'topic1/public_ip_addr', ... (116 bytes))
1678880086: sending puback to clientb (m4, rc0)
1678880086: sending publish to clienta (d0, q1, r0, m20726, 'topic1/public_ip_addr', ... (116 bytes))
1678880086: received puback from clienta (mid: 20726, rc:0)
... (lots of publish with puback from telemetry)
1678880086: sending publish to clienta (d0, q1, r0, m20742, 'topic1/swap', ... (169 bytes))
1678880086: received puback from clienta (mid: 20742, rc:0)
1678880086: received publish from clientb (d0, q1, r0, m21, 'topic1/syslog', ... (508219 bytes))
1678880086: sending puback to clientb (m21, rc0)
1678880086: sending publish to clienta (d0, q1, r0, m20743, 'topic1/syslog', ... (508219 bytes))
1678880086: received puback from clienta (mid: 20743, rc:0)
...
1678880088: received publish from clientb (d0, q1, r0, m22, 'topic1/version_ca_lockdown', ... (133 bytes))
1678880088: sending puback to clientb (m22, rc0)
... (here the services get restarted and afterwards the subscription seems broken for qos >= 1)
1678880118: received unsubscribe from clienta
1678880118:         telemetry/+
1678880118: clienta telemetry/+
1678880118: sending unsuback to clienta
1678880118: socket error on client clienta, disconnecting.
1678880119: received publish from clientb (d0, q1, r0, m136, 'topic1/release_id', ... (126 bytes))
1678880119: sending puback to clientb (m136, rc0)
1678880119: received publish from clientb (d0, q1, r0, m137, 'topic1/version_ca_configurator', ... (137 bytes))
1678880119: sending puback to clientb (m137, rc0)
...
1678880125: new connection from 127.0.0.1 on port 1883.
1678880125: new client connected from 127.0.0.1 as clienta (p2, c0, k10).
1678880125: no will message specified.
1678880125: sending connack to clienta (1, 0)
1678880125: sending publish to clienta (d1, q1, r0, m18734, 'topic1/matlab_version_expected', ... (152 bytes))
1678880125: sending publish to clienta (d1, q1, r0, m18735, 'topic1/coreagent_ota_state', ... (112 bytes))
... (again, lots of publish without puback)
1678880125: sending publish to clienta (d0, q1, r0, m20873, 'topic1/proc_led', ... (151 bytes))
1678880125: sending publish to clienta (d0, q1, r0, m20874, 'topic1/coreagent_ota_state', ... (112 bytes))
1678880125: sending publish to clienta (d0, q1, r0, m20875, 'topic1/ssid_wlan0', ... (99 bytes))
1678880125: received publish from clienta (d0, q2, r1, m1, 'conn-status', ... (7 bytes))
1678880125: sending pubrec to clienta (m1, rc0)
1678880125: received publish from clienta (d0, q2, r1, m2, 'topic1/telemetry_data_sent', ... (109 bytes))
1678880125: sending pubrec to clienta (m2, rc0)
1678880125: received pubrel from clienta (mid: 1)
1678880125: sending pubcomp to clienta (m1)
1678880125: received pubrel from clienta (mid: 2)
1678880125: sending pubcomp to clienta (m2)
1678880125: received subscribe from clienta
1678880125:         shutdown (qos 2)
1678880125: clienta 2 shutdown
1678880125: sending suback to clienta
1678880125: received publish from clienta (d0, q2, r1, m4, 'topic1/telemetry_queue_size', ... (119 bytes))
1678880125: sending pubrec to clienta (m4, rc0)
1678880125: received subscribe from clienta
1678880125:         telemetry/+ (qos 1)
1678880125: clienta 1 telemetry/+
1678880125: sending suback to clienta
1678880125: received pubrel from clienta (mid: 4)
1678880125: sending pubcomp to clienta (m4)
1678880125: received subscribe from clienta
1678880125:         telemetry-batch (qos 1)
1678880125: clienta 1 telemetry-batch
1678880125: sending suback to clienta
1678880125: received subscribe from clienta
1678880125:         topic1/# (qos 1)
1678880125: clienta 1 topic1/#
1678880125: sending suback to clienta
1678880126: received publish from clienta (d0, q2, r1, m8, 'conn-status', ... (6 bytes))
1678880126: sending pubrec to clienta (m8, rc0)
1678880126: received pubrel from clienta (mid: 8)
1678880126: sending pubcomp to clienta (m8)
...
1678880133: new connection from 127.0.0.1 on port 1883.
1678880133: new client connected from 127.0.0.1 as clientb (p2, c0, k10).
1678880133: no will message specified.
1678880133: sending connack to clientb (1, 0)
...
1678880135: received publish from clientb (d0, q1, r0, m3, 'topic1/provision_date', ... (131 bytes))
1678880135: sending puback to clientb (m3, rc0)
1678880135: received publish from clientb (d0, q1, r0, m4, 'topic1/processes', ... (99 bytes))
... (there are no more publish message from broker to clienta)
1678880135: sending puback to clientb (m20, rc0)
1678880135: received publish from clientb (d0, q1, r0, m21, 'topic1/net_response', ... (118 bytes))
1678880135: sending puback to clientb (m21, rc0)

我仍在研究一个最小的、可重现的示例，但不幸的是，到目前为止，我还没有在嵌入式环境之外重新创建错误。

编辑3：

为了完整起见，我设法构建了一个最小的、可重现的示例，并认为我也会发布它。

package main

import (
    "os"
    "os/signal"
    "syscall"
    "time"

    mqtt "github.com/eclipse/paho.mqtt.golang"
    "github.com/sirupsen/logrus"
)

func main() {
    c := make(chan os.Signal, 1)
    signal.Notify(c, os.Interrupt, syscall.SIGTERM)

    stop := make(chan bool, 1)
    optsA := mqtt.NewClientOptions().AddBroker("tcp://localhost:1883")
    logrus.Info("setup and connect clientA")
    clientA := setup(optsA, "clientA")
    if err := connect(clientA); err != nil {
        panic(err)
    }

    if tok := clientA.Subscribe("topic/+", 1, func(c mqtt.Client, m mqtt.Message) {
        logrus.Infof("received on topic %s ; message: %s", string(m.Topic()), string(m.Payload()))
    }); tok.Wait() && tok.Error() != nil {
        panic(tok.Error())
    }

    logrus.Infof("setup and connect clientB")
    optsB := mqtt.NewClientOptions().AddBroker("tcp://localhost:1883")
    clientB := setup(optsB, "clientB")
    if err := connect(clientB); err != nil {
        panic(err)
    }

    logrus.Info("start publishing from clientB")
    go publish(clientB, stop)

    logrus.Info("let clientA receive some messages from clientB")
    time.Sleep(500 * time.Millisecond)

    clientA.Disconnect(10)

    logrus.Info("wait until ClientB published more than max_inflight_messages")
    time.Sleep(1100 * time.Millisecond)

    logrus.Infof("connecting clientA again")
    clientA = setup(optsA, "clientA")
    if err := connect(clientA); err != nil {
        panic(err)
    }

    logrus.Info("wait shortly before subscribing")
    time.Sleep(1 * time.Second)

    logrus.Info("subscribe with clientA")
    if tok := clientA.Subscribe("topic/+", 1, func(c mqtt.Client, m mqtt.Message) {
        logrus.Infof("received on topic %s ; message: %s", string(m.Topic()), string(m.Payload()))
    }); tok.Wait() && tok.Error() != nil {
        panic(tok.Error())
    }

    <-c
    stop <- true
}

func publish(client mqtt.Client, stop chan bool) {
    for {
        select {
        case <-stop:
            return
        default:
            if tok := client.Publish("topic/exmaple", 1, false, "message"); tok.Wait() && tok.Error() != nil {
                logrus.WithError(tok.Error()).Warnf("failed to publish, continuing")
                continue
            }
            time.Sleep(50 * time.Millisecond)
        }
    }
}

func connect(client mqtt.Client) error {
    if tok := client.Connect(); tok.Wait() && tok.Error() != nil {
        logrus.WithError(tok.Error()).Error("failed to connect")
        return tok.Error()
    }
    return nil
}

func setup(opts *mqtt.ClientOptions, id string) mqtt.Client {
    opts.SetClientID(id)
    opts.SetOrderMatters(false)
    opts.SetCleanSession(false)
    // opts.SetDefaultPublishHandler(func(client mqtt.Client, msg mqtt.Message) {
    //  logrus.Infof("received message on topic %s which does not match any subscriptions (yet)", msg.Topic())
    // })
    opts.SetKeepAlive(10 * time.Second)
    opts.SetPingTimeout(1 * time.Second)
    opts.SetAutoReconnect(true)
    opts.SetConnectTimeout(15 * time.Second)

    cl := mqtt.NewClient(opts)
    return cl
}

正如 @brits 在他的回答中解释的那样，如果我取消注释 defaultpublishhandler，消息就会得到确认，并且订阅可以用于进一步的消息。

正确答案

感谢您的日志；事实上，mosquitto 没有收到 puback 在建立连接后立即收到的消息，这让我找到了可能的原因。

使用 mosquitto v1.6.x max_inflight_messages 默认为 10；因此，在 10 条未确认的消息之后，mosquitto 将不再发送任何消息。这就是它停止发送到 clienta 的原因。

如果没有处理程序，paho.mqtt.golang 将不会确认消息（如果启用日志记录，则在发生这种情况时会输出警告）。这样做的基本原理已经消失在时间的迷雾中（我添加了警告），但我怀疑这是因为，如果没有处理程序，则不能说该消息已被处理（因此不应被确认）。早期版本的 mosquitto 过去常常重新发送尚未确认的消息，但现在情况已不再如此（并且在 v5 规范中被禁止），这意味着它实际上是永久阻止。

就您而言，这两个因素相结合；您连接，接收 10 个 publish 数据包，然后订阅（设置处理程序），但此时 mosquitto 有 10 条消息正在发送，并且不会再发送。

修复方法是添加：

opts.SetDefaultPublishHandler(func(mqtt.Client, mqtt.Message) {})

这将添加一个默认的发布处理程序（忽略消息）；该处理程序存在的事实意味着 publish 数据包将被确认。

clienta 是否需要在离线状态下接收消息（它从 telemetry/+ 取消订阅，因此无论如何也不会收到这些消息）。如果没有，那么使用 opts.setcleansession(true) 是避免此问题的另一种方法。

如果您确实需要处理消息，请在连接之前使用 addroute 配置消息处理程序（我通常有一个包罗万象的 defaultpublishhandler ，它只记录消息，以便我可以看到遗漏了某些内容）。

到这里，我们也就讲完了《未收到其他客户端发送的消息的MQTT客户端》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！