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 的消息
總結
至此，我們完成了使用 paho.mqtt.golang 客戶端連接到 公共 MQTT 服務器，並實現了測試客戶端與 MQTT 服務器的連接、消息發布和訂閱。
接下來我們將會陸續發布更多關於物聯網開發及 MQTT 的相關文章，敬請關注。
版權聲明： 本文為 EMQ 原創，轉載請注明出處。