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

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

我在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＆ast; 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
...

Answer 2

您为什么不将邮件发送拆分为一群工作人员？