使用 GoRoutine 和通道在 Paho MQTT 上通过 websockets 发布消息的正确方法是什么？

在 Paho MQTT 库中使用 GoRoutine 和通道来发布消息时，需要采用正确的方法来确保消息的可靠性和有序性。本文将探讨两种可行的解决方案：一是通过取消对 `token.wait()` 的调用，实现异步消息发送；二是通过使用带缓冲通道的 GoRoutine，实现消息排序。本文还提供了示例代码，展示如何使用这些方法，以及如何将消息发送任务分成一组工作者，以提高性能。

作为我用来发布消息以进行测试的标准代码：

func main() {

    opts := mqtt.newclientoptions().addbroker("tcp://127.0.0.1:1883")
    opts.setclientid("myclientid_")
    opts.setdefaultpublishhandler(f)
    opts.setconnectionlosthandler(connlosthandler)

    opts.onconnect = func(c mqtt.client) {
        fmt.printf("client connected, subscribing to: test/topic\n")

        if token := c.subscribe("logs", 0, nil); token.wait() && token.error() != nil {
            fmt.println(token.error())
            os.exit(1)
        }
    }

    c := mqtt.newclient(opts)
    if token := c.connect(); token.wait() && token.error() != nil {
        panic(token.error())
    }


    for i := 0; i < 5; i++ {
        text := fmt.sprintf("this is msg #%d!", i)
        token := c.publish("logs", 0, false, text)
        token.wait()
    }

    time.sleep(3 * time.second)

    if token := c.unsubscribe("logs"); token.wait() && token.error() != nil {
        fmt.println(token.error())
        os.exit(1)
    }

    c.disconnect(250)
}

这个效果很好！但是在执行高延迟任务时大量传递消息，我的程序性能会很低，所以我必须使用 goroutine 和 channel。

所以，我正在寻找一种方法在 goroutine 中创建一个 worker，以便使用 golang 的 paho mqtt 库向浏览器发布消息，我很难找到满足我的需求的更好的解决方案，但经过一番搜索后，我找到了这段代码：

package main

import (
    "crypto/tls"
    "crypto/x509"
    "fmt"
    "io/ioutil"
    "strings"
    "time"

    MQTT "git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.golang.git"
    "linksmart.eu/lc/core/catalog"
    "linksmart.eu/lc/core/catalog/service"
)

// MQTTConnector provides MQTT protocol connectivity
type MQTTConnector struct {
    config        *MqttProtocol
    clientID      string
    client        *MQTT.Client
    pubCh         chan AgentResponse
    subCh         chan<- DataRequest
    pubTopics     map[string]string
    subTopicsRvsd map[string]string // store SUB topics "reversed" to optimize lookup in messageHandler
}

const defaultQoS = 1

func (c *MQTTConnector) start() {
    logger.Println("MQTTConnector.start()")

    if c.config.Discover && c.config.URL == "" {
        err := c.discoverBrokerEndpoint()
        if err != nil {
            logger.Println("MQTTConnector.start() failed to start publisher:", err.Error())
            return
        }
    }

    // configure the mqtt client
    c.configureMqttConnection()

    // start the connection routine
    logger.Printf("MQTTConnector.start() Will connect to the broker %v\n", c.config.URL)
    go c.connect(0)

    // start the publisher routine
    go c.publisher()
}

// reads outgoing messages from the pubCh und publishes them to the broker
func (c *MQTTConnector) publisher() {
    for resp := range c.pubCh {
        if !c.client.IsConnected() {
            logger.Println("MQTTConnector.publisher() got data while not connected to the broker. **discarded**")
            continue
        }
        if resp.IsError {
            logger.Println("MQTTConnector.publisher() data ERROR from agent manager:", string(resp.Payload))
            continue
        }
        topic := c.pubTopics[resp.ResourceId]
        c.client.Publish(topic, byte(defaultQoS), false, resp.Payload)
        // We dont' wait for confirmation from broker (avoid blocking here!)
        //<-r
        logger.Println("MQTTConnector.publisher() published to", topic)
    }
}


func (c *MQTTConnector) stop() {
    logger.Println("MQTTConnector.stop()")
    if c.client != nil && c.client.IsConnected() {
        c.client.Disconnect(500)
    }
}

func (c *MQTTConnector) connect(backOff int) {
    if c.client == nil {
        logger.Printf("MQTTConnector.connect() client is not configured")
        return
    }
    for {
        logger.Printf("MQTTConnector.connect() connecting to the broker %v, backOff: %v sec\n", c.config.URL, backOff)
        time.Sleep(time.Duration(backOff) * time.Second)
        if c.client.IsConnected() {
            break
        }
        token := c.client.Connect()
        token.Wait()
        if token.Error() == nil {
            break
        }
        logger.Printf("MQTTConnector.connect() failed to connect: %v\n", token.Error().Error())
        if backOff == 0 {
            backOff = 10
        } else if backOff <= 600 {
            backOff *= 2
        }
    }

    logger.Printf("MQTTConnector.connect() connected to the broker %v", c.config.URL)
    return
}

func (c *MQTTConnector) onConnected(client *MQTT.Client) {
    // subscribe if there is at least one resource with SUB in MQTT protocol is configured
    if len(c.subTopicsRvsd) > 0 {
        logger.Println("MQTTPulbisher.onConnected() will (re-)subscribe to all configured SUB topics")

        topicFilters := make(map[string]byte)
        for topic, _ := range c.subTopicsRvsd {
            logger.Printf("MQTTPulbisher.onConnected() will subscribe to topic %s", topic)
            topicFilters[topic] = defaultQoS
        }
        client.SubscribeMultiple(topicFilters, c.messageHandler)
    } else {
        logger.Println("MQTTPulbisher.onConnected() no resources with SUB configured")
    }
}

func (c *MQTTConnector) onConnectionLost(client *MQTT.Client, reason error) {
    logger.Println("MQTTPulbisher.onConnectionLost() lost connection to the broker: ", reason.Error())

    // Initialize a new client and reconnect
    c.configureMqttConnection()
    go c.connect(0)
}

func (c *MQTTConnector) configureMqttConnection() {
    connOpts := MQTT.NewClientOptions().
        AddBroker(c.config.URL).
        SetClientID(c.clientID).
        SetCleanSession(true).
        SetConnectionLostHandler(c.onConnectionLost).
        SetOnConnectHandler(c.onConnected).
        SetAutoReconnect(false) // we take care of re-connect ourselves

    // Username/password authentication
    if c.config.Username != "" && c.config.Password != "" {
        connOpts.SetUsername(c.config.Username)
        connOpts.SetPassword(c.config.Password)
    }

    // SSL/TLS
    if strings.HasPrefix(c.config.URL, "ssl") {
        tlsConfig := &tls.Config{}
        // Custom CA to auth broker with a self-signed certificate
        if c.config.CaFile != "" {
            caFile, err := ioutil.ReadFile(c.config.CaFile)
            if err != nil {
                logger.Printf("MQTTConnector.configureMqttConnection() ERROR: failed to read CA file %s:%s\n", c.config.CaFile, err.Error())
            } else {
                tlsConfig.RootCAs = x509.NewCertPool()
                ok := tlsConfig.RootCAs.AppendCertsFromPEM(caFile)
                if !ok {
                    logger.Printf("MQTTConnector.configureMqttConnection() ERROR: failed to parse CA certificate %s\n", c.config.CaFile)
                }
            }
        }
        // Certificate-based client authentication
        if c.config.CertFile != "" && c.config.KeyFile != "" {
            cert, err := tls.LoadX509KeyPair(c.config.CertFile, c.config.KeyFile)
            if err != nil {
                logger.Printf("MQTTConnector.configureMqttConnection() ERROR: failed to load client TLS credentials: %s\n",
                    err.Error())
            } else {
                tlsConfig.Certificates = []tls.Certificate{cert}
            }
        }

        connOpts.SetTLSConfig(tlsConfig)
    }

    c.client = MQTT.NewClient(connOpts)
}

这段代码正是我想要的！

但是作为 golang 中的菜鸟，我不知道如何在主函数中运行 start() 函数以及要传递什么参数！

特别是，我将如何使用通道将消息传递给工作人员（发布者）？！

我们将不胜感激您的帮助！

解决方案

我在 github repo 上发布了下面的答案，但由于您在这里提出了同样的问题，因此认为值得交叉发布（包含更多信息）。

当您说“在执行高延迟任务时大量传递消息”时，我假设您的意思是您想要异步发送消息（因此消息由与主代码运行不同的 go 例程处理） .

如果是这种情况，那么对您的初始示例进行非常简单的更改即可：

for i := 0; i < 5; i++ {
        text := fmt.sprintf("this is msg #%d!", i)
        token := c.publish("logs", 0, false, text)
        // comment out... token.wait()
    }

注意：您的示例代码可能会在消息实际发送之前退出；添加 time.sleep(10 * time.second) 会给它时间让它们熄灭；请参阅下面的代码以了解处理此问题的另一种方法

您的初始代码在消息发送之前停止的唯一原因是您调用了 token.wait()。如果您不关心错误（并且您不检查错误，所以我假设您不关心），那么调用 token.wait() 就没有什么意义（它只是等待消息发送；消息将消失无论您是否调用 token.wait()）。

如果您想记录任何错误，您可以使用以下内容：

for i := 0; i < 5; i++ {
        text := fmt.sprintf("this is msg #%d!", i)
        token := c.publish("logs", 0, false, text)
        go func(){
            token.wait()
            err := token.error()
            if err != nil {
                fmt.printf("error: %s\n", err.error()) // or whatever you want to do with your error
            }
        }()
    }

请注意，如果消息传递至关重要（但由于您没有检查错误，我假设它不是），您还需要做一些其他事情。

就您找到的代码而言；我怀疑这会增加您不需要的复杂性（并且需要更多信息才能解决此问题；例如，mqttprotocol 结构未在您粘贴的位中定义）。

额外一点...在您的评论中您提到“发布的消息必须排序”。如果这是必要的（因此您希望等到每条消息都已送达后再发送另一条消息），那么您需要类似以下内容：

msgchan := make(chan string, 200) // allow a queue of up to 200 messages
var wg sync.waitgroup
wg.add(1)
go func(){ // go routine to send messages from channel
    for msg := range msgchan {
        token := c.publish("logs", 2, false, msg) // use qos2 is order is vital
        token.wait()
        // should check for errors here
    }
    wg.done()
}()

for i := 0; i < 5; i++ {
        text := fmt.sprintf("this is msg #%d!", i)
        msgchan <- text
    }
close(msgchan) // this will stop the goroutine (when all messages processed)
wg.wait() // wait for all messages to be sent before exiting (may wait for ever is mqtt broker down!)

注意：这与 ilya kaznacheev 的解决方案类似（如果将workerpoolsize 设置为 1 并使通道缓冲）

正如您的评论表明等待组使这难以理解，这里是另一种可能更清晰的等待方式（等待组通常在您等待多个事物完成时使用；在本例中，我们只等待一个因此可以使用更简单的方法）

msgchan := make(chan string, 200) // allow a queue of up to 200 messages
done := make(chan struct{}) // channel used to indicate when go routine has finnished

go func(){ // go routine to send messages from channel
    for msg := range msgchan {
        token := c.publish("logs", 2, false, msg) // use qos2 is order is vital
        token.wait()
        // should check for errors here
    }
    close(done) // let main routine know we have finnished
}()

for i := 0; i < 5; i++ {
        text := fmt.sprintf("this is msg #%d!", i)
        msgchan <- text
    }
close(msgchan) // this will stop the goroutine (when all messages processed)
<-done // wait for publish go routine to complete

为什么不将消息发送分成一组工作人员呢？

类似这样的事情：

...
    const workerPoolSize = 10 // the number of workers you want to have
    wg := &sync.WaitGroup{}
    wCh := make(chan string)
    wg.Add(workerPoolSize) // you want to wait for 10 workers to finish the job

    // run workers in goroutines
    for i := 0; i < workerPoolSize; i++ {
        go func(wch <-chan string) {
            // get the data from the channel
            for text := range wch {
                c.Publish("logs", 0, false, text)
                token.Wait()
            }
            wg.Done() // worker says that he finishes the job
        }(wCh)
    }

    for i := 0; i < 5; i++ {
        // put the data to the channel
        wCh <- fmt.Sprintf("this is msg #%d!", i)
    }

        close(wCh)
    wg.Wait() // wait for all workers to finish
...

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《使用 GoRoutine 和通道在 Paho MQTT 上通过 websockets 发布消息的正确方法是什么？》文章吧，也可关注golang学习网公众号了解相关技术文章。