发布订阅 -- mqtt

mqtt 在线调试工具

http://www.emqx.io/online-mqtt-client

运行 emqx 服务（版本：5.0.26）

1. 使用 docker 运行服务：

    docker run --name emqx5026 \
        -p 1883:1883 \
        -p 8083:8083 \
        -p 8084:8084 \
        -p 8883:8883 \
        -p 18083:18083 \
        --restart=always \
        -d \
        emqx/emqx:5.0.26

2. 管理系统访问地址：

   xx.xx.xx.xx:18083
   admin
   public

使用 mqtt（版本：1.4.3）

1. 目录结构：

   |—— mqtt/   // 根目录
   ****|—— server/
   ********|—— main.go  // 发布者
   ****|—— client/
   ********|—— main.go  // 订阅者
   ****|—— go.mod  // go module 文件

2. mqtt/go.mod：

    module mqtt-demo
   
    require github.com/eclipse/paho.mqtt.golang v1.4.3

3. 运行订阅者：

   1. mqtt/client/main.go：

       package main
      
       import (
           "fmt"
           mqtt "github.com/eclipse/paho.mqtt.golang"
           "log"
           "os"
       )
      
       const (
           address = "tcp://xx.xx.xx.xx:1883" // mqtt 协议
           //address = "ws://xx.xx.xx.xx:8083/mqtt" // websocket 协议
           topic = "test/1"
           //topic = "test/#"  // "#"批量订阅
           clientId = "2" // 订阅者 client-id，取固定值
       )
      
       func main() {
           mqtt.ERROR = log.New(os.Stdout, "error--", 0)
           opts := mqtt.NewClientOptions().AddBroker(address).SetCleanSession(false).SetClientID(clientId)
           client := mqtt.NewClient(opts)
           if token := client.Connect(); token.Wait() && token.Error() != nil {
               panic(token.Error())
           }
      
           client.Subscribe(topic, 1, func(client mqtt.Client, message mqtt.Message) {
               fmt.Printf("topic: %s, message: %s \n", message.Topic(), string(message.Payload()))
           })
      
           select {}
       }

   2. 运行：

      cd mqtt/client
      go run main.go

4. 运行发布者：

   1. mqtt/server/main.go：

       package main
      
       import (
           "fmt"
           mqtt "github.com/eclipse/paho.mqtt.golang"
           "log"
           "math/rand"
           "os"
           "time"
       )
      
       const (
           address = "tcp://xx.xx.xx.xx:1883" // mqtt 协议
           //address = "ws://xx.xx.xx.xx:8083/mqtt" // websocket 协议
           topic = "test/1"
       )
      
       func RandStr(length int) string {
           str := "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
           bytes := []byte(str)
           result := []byte{}
           rand.Seed(time.Now().UnixNano() + int64(rand.Intn(100)))
           for i := 0; i < length; i++ {
               result = append(result, bytes[rand.Intn(len(bytes))])
           }
           return string(result)
       }
      
       func main() {
           mqtt.ERROR = log.New(os.Stdout, "error--", 0)
           clientId := RandStr(16) // 发布者 client-id，取随机值
           opts := mqtt.NewClientOptions().AddBroker(address).SetCleanSession(false).SetClientID(clientId)
           client := mqtt.NewClient(opts)
           if token := client.Connect(); token.Wait() && token.Error() != nil {
               panic(token.Error())
           }
      
           msg := "msg from server"
           client.Publish(topic, 1, false, msg)
      
           client.Disconnect(250)
           fmt.Println("success!")
       }

   2. 运行：

      cd mqtt/server
      go run main.go