Golang 中MQTT如何使用

這篇文章給大家介紹Golang 中MQTT如何使用，內容非常詳細，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。

Golang 是 Google 開發的一種靜態強類型、編譯型、并發型，并具有垃圾回收功能的編程語言。Go 的表現力強、簡潔、干凈、高效。它的并發機制使它能夠輕松地編寫程序，從而最大限度地利用多核和網絡機器，而它新穎的類型系統則能夠實現靈活和模塊化的程序構造。Go 快速編譯成機器代碼，但又具有垃圾回收的便利性和運行時反射的強大功能。它是一種快速的、靜態類型化的、編譯后的語言，就像一種動態類型化的、解釋的語言。

MQTT 是一種基于發布/訂閱模式的 輕量級物聯網消息傳輸協議 ，可以用極少的代碼和帶寬為聯網設備提供實時可靠的消息服務，它廣泛應用于物聯網、移動互聯網、智能硬件、車聯網、電力能源等行業。

本文主要介紹如何在 Golang 項目中使用 paho.mqtt.golang 客戶端庫 ，實現客戶端與 MQTT 服務器 的連接、訂閱、收發消息等功能。

項目初始化

本項目基于 go1.13.12 進行開發測試

go version
go 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 main

import (
	"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 = 8883
opts := 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 main

import (
	"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 的消息

關于Golang 中MQTT如何使用就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。