Golang 是 Google 开发的一种动态强类型、编译型、并发型，并具备垃圾回收性能的编程语言。Go 的表现力强、简洁、洁净、高效。它的并发机制使它可能轻松地编写程序，从而最大限度地利用多核和网络机器，而它新鲜的类型零碎则可能实现灵便和模块化的程序结构。Go 疾速编译成机器代码，但又具备垃圾回收的便利性和运行时反射的弱小性能。它是一种疾速的、动态类型化的、编译后的语言，就像一种动静类型化的、解释的语言。

MQTT 是一种基于公布/订阅模式的 轻量级物联网音讯传输协定 ，能够用极少的代码和带宽为联网设施提供实时牢靠的音讯服务，它广泛应用于物联网、挪动互联网、智能硬件、车联网、电力能源等行业。

本文次要介绍如何在 Golang 我的项目中应用 paho.mqtt.golang 客户端库 ，实现客户端与 MQTT 服务器 的连贯、订阅、收发音讯等性能。

我的项目初始化

本我的项目基于 go1.13.12 进行开发测试

go versiongo version go1.13.12 darwin/amd64

本我的项目应用 paho.mqtt.golang 作为 MQTT 客户端库，装置：

go get github.com/eclipse/paho.mqtt.golang

Go MQTT 应用

本文将应用 EMQ X 提供的 收费公共 MQTT 服务器，该服务基于 EMQ X 的 MQTT 物联网云平台 创立。服务器接入信息如下：

  • Broker: broker.emqx.io
  • TCP Port: 1883
  • Websocket Port: 8083

连贯 MQTT 服务器

package mainimport (    "fmt"    mqtt "github.com/eclipse/paho.mqtt.golang"    "time")var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {    fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())}var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {    fmt.Println("Connected")}var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {    fmt.Printf("Connect lost: %v", err)}func main() {    var broker = "broker.emqx.io"    var port = 1883    opts := mqtt.NewClientOptions()    opts.AddBroker(fmt.Sprintf("tcp://%s:%d", broker, port))    opts.SetClientID("go_mqtt_client")    opts.SetUsername("emqx")    opts.SetPassword("public")    opts.SetDefaultPublishHandler(messagePubHandler)    opts.OnConnect = connectHandler    opts.OnConnectionLost = connectLostHandler    client := mqtt.NewClient(opts)    if token := client.Connect(); token.Wait() && token.Error() != nil {        panic(token.Error())  }}
  • ClientOptions：用于设置 broker，端口，客户端 id ，用户名明码等选项
  • messagePubHandler：全局 MQTT pub 音讯解决
  • connectHandler：连贯的回调
  • connectLostHandler：连贯失落的回调

如果想应用 TLS 连贯，能够如下设置：

func NewTlsConfig() *tls.Config {    certpool := x509.NewCertPool()    ca, err := ioutil.ReadFile("ca.pem")    if err != nil {        log.Fatalln(err.Error())    }    certpool.AppendCertsFromPEM(ca)    // Import client certificate/key pair    clientKeyPair, err := tls.LoadX509KeyPair("client-crt.pem", "client-key.pem")    if err != nil {        panic(err)    }    return &tls.Config{        RootCAs: certpool,        ClientAuth: tls.NoClientCert,        ClientCAs: nil,        InsecureSkipVerify: true,        Certificates: []tls.Certificate{clientKeyPair},    }}

如果不设置客户端证书，能够如下设置：

func NewTlsConfig() *tls.Config {    certpool := x509.NewCertPool()    ca, err := ioutil.ReadFile("ca.pem")    if err != nil {        log.Fatalln(err.Error())    }    certpool.AppendCertsFromPEM(ca)    return &tls.Config{        RootCAs: certpool,}

而后设置 TLS

var broker = "broker.emqx.io"var port = 8883opts := mqtt.NewClientOptions()opts.AddBroker(fmt.Sprintf("tcp://%s:%d", broker, port))tlsConfig := NewTlsConfig()opts.SetTLSConfig(tlsConfig)// other options

订阅

func sub(client mqtt.Client) {    topic := "topic/test"    token := client.Subscribe(topic, 1, nil)    token.Wait()    fmt.Printf("Subscribed to topic %s", topic)}

公布音讯

func publish(client mqtt.Client) {    num := 10    for i := 0; i < num; i++ {        text := fmt.Sprintf("Message %d", i)        token := client.Publish("topic/test", 0, false, text)        token.Wait()        time.Sleep(time.Second)    }}

测试

咱们应用以下代码进行测试

package mainimport (    "fmt"    mqtt "github.com/eclipse/paho.mqtt.golang"    "log"    "time")var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {    fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())}var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {    fmt.Println("Connected")}var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {    fmt.Printf("Connect lost: %v", err)}func main() {    var broker = "broker.emqx.io"    var port = 1883    opts := mqtt.NewClientOptions()    opts.AddBroker(fmt.Sprintf("tcp://%s:%d", broker, port))    opts.SetClientID("go_mqtt_client")    opts.SetUsername("emqx")    opts.SetPassword("public")    opts.SetDefaultPublishHandler(messagePubHandler)    opts.OnConnect = connectHandler    opts.OnConnectionLost = connectLostHandler    client := mqtt.NewClient(opts)    if token := client.Connect(); token.Wait() && token.Error() != nil {        panic(token.Error())    }    sub(client)    publish(client)    client.Disconnect(250)}func publish(client mqtt.Client) {    num := 10    for i := 0; i < num; i++ {        text := fmt.Sprintf("Message %d", i)        token := client.Publish("topic/test", 0, false, text)        token.Wait()        time.Sleep(time.Second)    }}func sub(client mqtt.Client) {    topic := "topic/test"    token := client.Subscribe(topic, 1, nil)    token.Wait()  fmt.Printf("Subscribed to topic: %s", topic)}

运行代码，能够看到 MQTT 连贯、订阅胜利，并能胜利收到订阅 topic 的音讯

总结

至此，咱们实现了应用 paho.mqtt.golang 客户端连贯到 公共 MQTT 服务器，并实现了测试客户端与 MQTT 服务器的连贯、音讯公布和订阅。

接下来咱们将会陆续公布更多对于物联网开发及 MQTT 的相干文章，敬请关注。

版权申明： 本文为 EMQ 原创，转载请注明出处。

原文链接：https://www.emqx.io/cn/blog/how-to-use-mqtt-in-golang