重新初始化后，Golang-Paho MQTT 订阅者频繁断开并报错 EOF

一分耕耘，一分收获！既然打开了这篇文章《重新初始化后，Golang-Paho MQTT 订阅者频繁断开并报错 EOF》，就坚持看下去吧！文中内容包含等等知识点...希望你能在阅读本文后，能真真实实学到知识或者帮你解决心中的疑惑，也欢迎大佬或者新人朋友们多留言评论，多给建议！谢谢！

我正在尝试动态更改 mqtt 客户端处理程序和证书，这会在订阅者和发布者连接时导致订阅者 eof

这就是我想做的事情

1] 我正在初始化订阅者/发布者（使用firstpubhandler、firstconnecthandler 和默认证书）

2]使用发布者在服务器上发送注册消息以获取新证书详细信息

3] 服务器将返回证书详细信息，该响应将由主题 .../id/certificate 上的firstconnecthandler 处理以下载证书。

4]firstpubhandler 将处理服务器的响应并重新初始化发布者/订阅者（使用 messagepubhandler、connecthandler 和新下载的证书），connecthandler 将侦听所有主题/id/+

一切正常，除了当我重新初始化订阅者/发布者时，订阅者不断断开连接并出现错误“eof”

我在这里做错了什么吗？或者有没有更好的方法来实现这一点？ 感谢任何帮助

--主要功能

var opt params
var publisher mqtt.client
var subscriber mqtt.client

func main() {
    initializebroker(firstpubhandler, firstconnecthandler)

    //ultimately it will trigger message on ".../id/certificate" topic which will be handled byfirstconnecthandler 
    publishregistrationmessage(publisher)

    c := make(chan os.signal, 1)
    signal.notify(c, os.interrupt, syscall.sigterm)
    done := make(chan os.signal, 1)
    signal.notify(done, os.interrupt, syscall.sigterm)
    go func() {
        for {
        }
    }()
    <-done
    <-c
    disconnectbrocker()
}

-- 处理程序

// first handlers
var firstpubhandler mqtt.messagehandler = func(client mqtt.client, msg mqtt.message) {
    downloadcertificates(msg.payload())
    initializebroker(messagepubhandler, connecthandler)
}

var firstconnecthandler mqtt.onconnecthandler = func(c mqtt.client) {
    if token := c.subscribe(opt.subclientid+"/id/certificate", 0, firstpubhandler); token.wait() && token.error() != nil {
        log.error(token.error())
    }
}

// second handlers
var messagepubhandler mqtt.messagehandler = func(client mqtt.client, msg mqtt.message) {
    processmessage(msg.payload())
}

var connecthandler mqtt.onconnecthandler = func(c mqtt.client) {
    if token := c.subscribe(opt.subclientid+"/id/+", 0, messagepubhandler); token.wait() && token.error() != nil {
        log.error(token.error())
    }
}

// common handler
var connectlosthandler mqtt.connectionlosthandler = func(client mqtt.client, err error) {
    log.info(err)
}

-- mqtt 代理初始化

func InitializeBroker(lMessageHandler mqtt.MessageHandler, lConnectHandler mqtt.OnConnectHandler) {
    statusPublishTopic := opt.PubClientId/id
    nodeSubscribeTopic := opt.SubClientId/id

    // Build the options for the publish client
    publisherOptions := mqtt.NewClientOptions()
    publisherOptions.AddBroker(opt.Broker)
    publisherOptions.SetClientID(statusPublishTopic)
    publisherOptions.SetDefaultPublishHandler(lMessageHandler)
    publisherOptions.OnConnectionLost = connectLostHandler

    // Build the options for the subscribe client
    subscriberOptions := mqtt.NewClientOptions()
    subscriberOptions.AddBroker(opt.Broker)
    subscriberOptions.SetClientID(nodeSubscribeTopic)
    subscriberOptions.SetDefaultPublishHandler(lMessageHandler)
    subscriberOptions.OnConnectionLost = connectLostHandler
    subscriberOptions.OnConnect = lConnectHandler

    if !opt.NoTLS {
        tlsconfig, err := NewTLSConfig()
        if err != nil {
            log.Fatalf(err)
        }

        subscriberOptions.SetTLSConfig(tlsconfig)
        publisherOptions.SetTLSConfig(tlsconfig)
    }

    publisher = mqtt.NewClient(publisherOptions)
    if token := publisher.Connect(); token.Wait() && token.Error() != nil {
        log.Fatalf(token.Error())
    }

    subscriber = mqtt.NewClient(subscriberOptions)
    if token := subscriber.Connect(); token.Wait() && token.Error() != nil {
        log.Fatalf(token.Error())
    }
}

func NewTLSConfig() (config *tls.Config, err error) {
    certpool := x509.NewCertPool()
    pemCerts, err := ioutil.ReadFile(rootCert)
    if err != nil {
        return nil, err
    }
    certpool.AppendCertsFromPEM(pemCerts)

    cert, err := tls.LoadX509KeyPair(nodeCertFilePath, pvtKeyFilePath)
    if err != nil {
        return nil, err
    }

    config = &tls.Config{
        RootCAs:      certpool,
        ClientAuth:   tls.NoClientCert,
        ClientCAs:    nil,
        Certificates: []tls.Certificate{cert},
    }
    return config, nil
}

正确答案

根据对您的代码的快速审查，我相信这是正在发生的事情（因为您没有提供所有代码，因此涉及一些猜测）：

1. main() 调用 InitializeBroker，这会创建与代理的两个连接。默认发布处理程序设置为 firstPubHandler ，并在 OnConnect 处理程序中订阅 SubClientId+"/id/Certificate
2. 收到消息 (firstPubHandler) 后，您可以从消息中获取一个证书，并使用它通过相同的客户端 ID 但不同的 OnConnect/默认发布处理程序来建立一组与代理的新连接。

因此，在第 2 点之后，您实际上与代理有两组独立的连接（总共 4 个连接）。然而 MQTT-3.1.4-2 （参见 spec）指出：

如果 ClientId 表示客户端已连接到服务器，则服务器必须断开现有客户端的连接。

因此，当建立第二组连接时，代理将删除第一组连接。这就是您看到的“EOF”断开连接。第二组连接仍将保持正常状态。由于您对第一组和第二组连接使用相同的 connectLostHandler，因此您无法在日志中看到哪个连接正在终止。

总之，我相信您的代码确实有效。但是，您可能应该在 firstConnectHandler 中调用 c.Disconnect() ，以便在建立第二组连接之前完全关闭初始连接。您还需要将 publisher 存储在某处，以便您可以同时断开该连接。

注意：我很难理解你为什么要这样做。建立初始连接来检索证书似乎会降低系统的整体安全性。标准方法是为每个客户端提供唯一的证书，然后在代理上使用 ACL 来应用任何必要的限制。对于许多代理，您可以在 ACL 中使用证书通用名称（从而无需第二个连接）。

终于介绍完啦！小伙伴们，这篇关于《重新初始化后，Golang-Paho MQTT 订阅者频繁断开并报错 EOF》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布Golang相关知识，快来关注吧！