Golang MQTT 发布和订阅

Question

有人知道我在哪里可以获取一些示例MQTT客户端Go（golang）代码，该代码可以在无限循环中执行发布和订阅操作吗？

我正在使用MacOS上运行的Mosquitto代理进行消息传递。

更详细地说...

1. 从网络中获取一条消息（主题）
2. 根据该消息计算某些内容
3. 将计算结果发送回网络（主题）

这是我正在使用的代码:

package main

import (
"fmt"
 MQTT "github.com/eclipse/paho.mqtt.golang"
"os"
"time"
)

var knt int

var f MQTT.MessageHandler = func(client MQTT.Client, msg MQTT.Message) 
{ 
   fmt.Printf("MSG: %s\n", msg.Payload())
   text:= fmt.Sprintf("this is result msg #%d!", knt)
   knt++
   token := client.Publish("nn/result", 0, false, text)
   token.Wait()
}

func main() {
   knt = 0

   opts := MQTT.NewClientOptions().AddBroker("tcp://localhost:1883")
   opts.SetClientID("mac-go")
   opts.SetDefaultPublishHandler(f)

   c := MQTT.NewClient(opts)
   if token := c.Connect(); token.Wait() && token.Error() != nil {
        panic(token.Error())
   }

  if token := c.Subscribe("nn/sensors", 0, nil); token.Wait() && 
     token.Error() != nil {
     fmt.Println(token.Error())
     os.Exit(1)
  }

  time.Sleep(3 * time.Second)
} //end of main

我查看了Go文档，希望能找到一些提示来保持连接的开放状态，但是似乎没有相关的内容。我可以在“subscribe”上进行无限循环，但这似乎效率低下。

我查阅了Go文档，但并未找到与保持连接开放有关的有用信息。虽然可以通过在“subscribe”上进行无限循环来实现，但效率不高。

- Phillip Neal

---

在 Stack Overflow 上询问教程是不合适的，展示你已经尝试过的内容，有人会帮助你解决问题。 - hardillb

好的。谢谢反馈。我找到了一个解决方案……我想。我把解决方案代码添加到问题中。它在测试环境中运行。现在我要在真实环境中尝试一下。 - Phillip Neal

Answer 1

我查看了GoDocs以获取如何保持连接的一些提示，但似乎没有相关内容。我可以在“subscribe”上做一个无限循环，但那似乎效率低下。

好的，在这里找到了解决方案。基本上，我必须为订阅打开一个通道。以下是新代码：

package main

import (
    "fmt"
    MQTT "github.com/eclipse/paho.mqtt.golang"
    "os"
    "os/signal"
    "syscall"
)

var knt int
var f MQTT.MessageHandler = func(client MQTT.Client, msg MQTT.Message) {
    fmt.Printf("MSG: %s\n", msg.Payload())
    text := fmt.Sprintf("this is result msg #%d!", knt)
    knt++
    token := client.Publish("nn/result", 0, false, text)
    token.Wait()
}

func main() {
    knt = 0
    c := make(chan os.Signal, 1)
    signal.Notify(c, os.Interrupt, syscall.SIGTERM)

    opts := MQTT.NewClientOptions().AddBroker("tcp://localhost:1883")
    opts.SetClientID("mac-go")
    opts.SetDefaultPublishHandler(f)
    topic := "nn/sensors"

    opts.OnConnect = func(c MQTT.Client) {
            if token := c.Subscribe(topic, 0, f); token.Wait() && token.Error() != nil {
                    panic(token.Error())
            }
    }
    client := MQTT.NewClient(opts)
    if token := client.Connect(); token.Wait() && token.Error() != nil {
            panic(token.Error())
    } else {
            fmt.Printf("Connected to server\n")
    }
    <-c
}

- Phillip Neal

---

我编辑了我的问题，并将我的答案移动到答案部分。 - Phillip Neal

MQTT "github.com/eclipse/paho.mqtt.golang" 的理由是什么？如果有的话，为什么不是这个： FMT "fmt" - Rick O'Shea