使用此代码（Paho MQTT）作为 GoRoutine 并通过通道传递消息以通过 websockets 发布的正确方法是什么

Question

作为标准代码，我使用发布消息进行测试：

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() 在我的主函数中运行以及传递什么参数！

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

我们将不胜感激！

Answer 1

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

像这样：

...
    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
...

Answer 2

我在 github repo 上发布了下面的答案，但正如你在这里问过同样的问题一样，我认为值得交叉发布（提供更多信息）。

当你说 "passing messages in mass while doing high latency tasks" 时，我假设你的意思是你想异步发送消息（因此消息由不同的 go-routine 处理，而不是你的主代码是运行 on） .

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

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 结构）。

额外的一点...在您的评论中您提到了 "Published messages must be ordered"。如果那是必要的（所以你想等到每条消息都发送完再发送另一条消息）那么你需要这样的东西：

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

使用此代码（Paho MQTT）作为 GoRoutine 并通过通道传递消息以通过 websockets 发布的正确方法是什么

What is the right way to use this code (Paho MQTT) as GoRoutine and pass messages via channel to publish via websockets

go

mqtt

paho