蚊子的持久性不一致

Question

我在 mosquitto 上看到消息持久性和 qos=2 的消息传递不一致。有什么我做错了吗？

我有一个简单的测试应用程序，它使用 clientId="receive-client" 注册消费主题，但立即断开连接。然后它以 clientId="send-client" 的身份连接并发布 10 条消息，“message #1”...“message #10”。然后断开连接，等待五秒钟，然后在打印和计算收到的消息时再次连接以使用“receive-client”消费。

结果不一致。有时我收到 6 条消息，有时是 8 条。典型的输出是这样的：

WARN[0005] GOT A MESSAGE:message #1                     

WARN[0005] GOT A MESSAGE:message #2                     

WARN[0005] GOT A MESSAGE:message #3                     

WARN[0005] GOT A MESSAGE:message #4                     

WARN[0005] GOT A MESSAGE:message #5                     

WARN[0005] GOT A MESSAGE:message #6                     

WARN[0005] GOT A MESSAGE:message #7                     

WARN[0005] GOT A MESSAGE:message #8                     

WARN[0305] PAUSE                                        

WARN[0605] received message count=8                     

我的版本信息是 1.4.15。我的 mosquitto.conf 是：

pid_file /var/run/mosquitto.pid

persistence true

persistence_location /var/lib/mosquitto/

allow_anonymous false

password_file /etc/mosquitto/passwd

log_dest file /var/log/mosquitto/mosquitto.log

最初 /var/lib/mosquitto/mosquitto.db 直到运行了几次迭代才会出现。我的测试应用在这里：

import (

    mqtt "github.com/eclipse/paho.mqtt.golang"

    log "github.com/sirupsen/logrus"

    "time"

)

var receivedMsg int

func Persist() {

    const TOPIC = "test"

    const URL = "tcp://localhost:1883"

    const USERNAME = "myuser"

    const PASSWORD = "mypassword"

    defer printReceived()

    options := mqtt.NewClientOptions().AddBroker(URL).SetUsername(USERNAME).SetPassword(PASSWORD)

    options.SetCleanSession(false)

    options.SetConnectRetry(true)

    options.SetConnectRetryInterval(10 * time.Millisecond)

Answer 1

在 QOS 2 上发布是一个多步骤过程，因此最可能的原因是您在所有消息真正完成发布到代理之前断开了发布客户端的连接。您可能应该在循环中进行发布，并使用调用返回的令牌client.publish()来等待它完成，然后再断开客户端。例如，如示例所示：//Publish 5 messages to /go-mqtt/sample at qos 1 and wait for the receipt//from the server after sending each messagefor i := 0; i < 5; i++ {&nbsp; text := fmt.Sprintf("this is msg #%d!", i)&nbsp; token := c.Publish("go-mqtt/sample", 0, false, text)&nbsp; token.Wait()}