MQTT:輕量的消息隊列協定

這幾年隨著物聯網(IoT)越喊越熱,MQTT 這套通訊協定也越來越常聽人提起。物聯網的核心精神是將終端數據傳輸到網路,網路中的運算單元在分析這些數據後,可以將它轉化為使用者想要的應用。在網路中,數據是用通訊協定來交換,而常見的通訊協定 HTTP 結構相對複雜,傳輸成本較高,也更要求終端裝置的效能,而且 HTTP 採用輪詢(Polling)機制來取得資料,需要客戶端頻繁跟裝置拉取訊息,不適合 IoT 的應用;相對的,MQTT 輕量、採用發佈/訂閱模式,更適合 IoT 的傳輸需求。

最常見的 MQTT 工具是由 Eclipse Foundation 維護的 mosquitto,相關工具也都有開源。本文會講解如何使用 Golang 的 mqtt 套件,搭配 mosquitto 的中間人 Broker 服務,走一次 MQTT 的使用。

Prepare Environment

首先要來準備 MQTT 的環境,在實際應用中,為了維護方便,會建議 MQTT 的伺服端架在 Linux 上,客戶端則看具體需求。因為我們只是要驗證概念,就選擇使用 Windows + WSL 作為開發環境,來對應 GUI + Server 配置兩種需要。

WSL 的 mosquitto 安裝方式跟 Debian 相同

sudo apt install mosquitto mosquitto-clients

安裝後,開啟服務

mosquitto

這樣 MQTT 的伺服端就架好了。接著在 Windows 上安裝 MQTT 客戶端,如果習慣使用 GUI 的話,可以用 MQTT.fx

choco install mqttfx -y

開啟 MQTT.fx,設定連線的資料。

由於伺服器開在本機,Broker 輸入 127.0.0.1;Port 用預設的 1883;Client ID 主要供 Broker 判斷 Client 用,不要重複即可,這邊用亂數。

在 Subscribe 的 Tab 中,可以選擇要訂閱的訊息,MQTT 是採用發佈/訂閱架構,Broker 收到發佈者的消息後,會將消息傳送給所有訂閱者。我們可以用通配符 # 來訂閱所有主題。

到 Publish 的 Tab,試著發佈一則訊息訊息

回到 Subscribe,可以看到該訊息已經被接收

有第三方的環境後,就可以拿來驗證自己開發的應用程式是否正確。

Send MQTT Message

建置完環境,開始來開發應用程式,同樣使用前面的專案架構,建立 cmd 並在 pkg 下加入需要的檔案

project
├── cmd
│   └── mqtt
│       └── main.go
├── configs
├── pkg
│   └── mqtt
│       ├── message.go
│       └── mqtt.go
├── go.mod
└── README.md

pkg/mqtt/message.go 用來建立訊息的模型;pkg/mqtt/mqtt.go 用來處理 MQTT 客戶端。

引入需要的 package

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

一個一個來看,pkg/mqtt/message.go 中的訊息格式是

package mqtt

type Message struct {
    Header  string
    Content string
    Footer  string
}

將訊息分為三部分,有 Header、Content、Footer,都是字串。當然用別的格式也行,就看各自需求。

接著 pkg/mqtt/mqtt.go 的內容是

package mqtt

import (
    "encoding/json"
    "fmt"
    "log"
    "math/rand"
    "os"
    "strconv"
    "time"

    mqtt "github.com/eclipse/paho.mqtt.golang"
)

// Client is a mqtt client
type Client struct {
    client mqtt.Client
}

// New mqtt client and connect to broker
func New() Client {
    c := Client{}
    mqtt.DEBUG = log.New(os.Stdout, "", 0)
    mqtt.ERROR = log.New(os.Stdout, "", 0)
    s1 := rand.NewSource(time.Now().UnixNano())
    r1 := rand.New(s1)
    clientID := strconv.Itoa(r1.Int())
    opts := mqtt.NewClientOptions().AddBroker("tcp://127.0.0.1:1883").SetClientID(clientID)

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

    return c
}

// Publish mqtt message
func (m Client) Publish(topic string, payload interface{}) {
    text, _ := json.Marshal(payload)
    token := m.client.Publish(topic, 0, false, text)
    token.Wait()
}

前面宣告一個結構,該結構內有 MQTT 的相關方法,可供應用端調用。

// Client is a mqtt client
type Client struct {
    client mqtt.Client
}

建立一個 New function 將此結構回傳

func New() Client {
    c := Client{}
    mqtt.DEBUG = log.New(os.Stdout, "", 0)
    mqtt.ERROR = log.New(os.Stdout, "", 0)

3、4 兩行的意思是將 MQTT 客戶端的 log 輸出到標準 log 中,方便 Debug。

亂數產生一組 Client ID,並設定 Broker 的 Address,類似前面 MQTT.fx 的設定

    s1 := rand.NewSource(time.Now().UnixNano())
    r1 := rand.New(s1)
    clientID := strconv.Itoa(r1.Int())
    opts := mqtt.NewClientOptions().AddBroker("tcp://127.0.0.1:1883").SetClientID(clientID)

新建一個 MQTT client,並指給 Client 中的 client;使用 Connect 來連線,並用 Wait 跟 Error 來確認連線成功,沒出現問題

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

還要記得加入 Publish,用來發佈訊息

// Publish mqtt message
func (m Client) Publish(topic string, payload interface{}) {
    text, _ := json.Marshal(payload)
    token := m.client.Publish(topic, 0, false, text)
    token.Wait()
}

Publish 需要兩個參數,分別是 Topic 跟 Payload,可以合理預期進來的 Payload 是 Message 格式,因此用 json.Marshal 將它序列化,方便傳輸。

好的,pkg 開發完成,回到應用端,修改 cmd/mqtt/main.go 的內容為

package main

import (
    "example/pkg/mqtt"
    "time"
)

func main() {
    client := mqtt.New()
    payload := mqtt.Message{
        Header: "this is header",
    }
    client.Publish("topic/golang", payload)
}

設定的 Topic 是 “topic/golang”,Payload 的 Header 帶有內容;Content 跟 Footer 為空。

編譯並執行,可以看到

ken@DESKTOP-2R08VK6:~/git/golang-medium-example$ ./bin/mqtt 
6966058732451841891
[client]   Connect()
[store]    memorystore initialized
[client]   about to write new connect msg
[client]   socket connected to broker
[client]   Using MQTT 3.1.1 protocol
[net]      connect started
[net]      received connack
[client]   client is connected
[net]      logic started
[net]      logic waiting for msg on ibound
[client]   exit startClient
[net]      outgoing started
[net]      outgoing waiting for an outbound message
[net]      incoming started
[pinger]   keepalive starting
[client]   enter Publish
[client]   sending publish message, topic: topic/golang
[net]      obound wrote msg, id: 0
[net]      outgoing waiting for an outbound message

同時,MQTT.fx 也能看到訂閱的訊息

Receive MQTT Message

如果是對純收集資訊的終端應用,Publish 應該已經能滿足需求了,但有時會需要更細緻的互動,例如出門在外,希望透過 MQTT 發佈訊息,將家裡的電燈關掉,這時就會要求終端需要具備 Subscribe 的能力。

作法很單純,修改 pkg/mqtt/mqtt.go,加入 Subscribe

// Subscribe mqtt message
func (m Client) Subscribe(topic string) {
    if token := m.client.Subscribe(topic, 0, nil); token.Wait() && token.Error() != nil {
        fmt.Println(token.Error())
        os.Exit(1)
    }
}

Client 已經可以訂閱主題了,但收到主題後希望進行什麼處理呢?因此加入 Handler,在收到訊息後,將 Topic 跟 Payload 印出,並將這個 Handler 設為預設的處理方式

var f mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
    fmt.Printf("TOPIC: %s\n", msg.Topic())
    fmt.Printf("MSG: %s\n", msg.Payload())
}

// ...

func New() Client {
    // ...
    opts.DefaultPublishHandler = f
    // ...
}

應用端也跟著修改

func main() {
    client := mqtt.New()
    payload := mqtt.Message{
        Header: "this is header",
    }
    client.Publish("topic/golang", payload)
    client.Subscribe("topic/golang")
    time.Sleep(30 * time.Second)
}

後面加個 Sleep 30 秒,讓應用程式不要立刻結束,才能看到訂閱的訊息。

同樣編譯並執行,在程式執行期間,使用 MQTT.fx 發佈訊息

在 Golang 可以看到對應的消息印出

[net]      received publish, msgId: 0
[net]      putting msg on onPubChan
[net]      done putting msg on incomingPubChan
[net]      logic waiting for msg on ibound
TOPIC: topic/golang
MSG: hello
[pinger]   ping check 20.0000549

Encrypt Message

如果訊息的傳輸是在內網,使用未加密的傳輸就行了,但如果訊息會通過公有網路進到伺服器,就需要對訊息進行加密了。TLS 是常用的安全協定,只要讓 MQTT 跑在 TLS 上,MQTT 的訊息就會被加密,即使第三方透過網路竊聽,也無法取得訊息內容。

現代的訊息加密都是透過信任鍊的方式,具體可以參照這篇的說明,不再贅述。為加密訊息,要先有 CA 簽證,因為是自己開發的東西,使用自簽 CA 來做為信任鍊的基礎。

先產生一組 CA 用的私鑰 ca.key

openssl genrsa -out configs/mqtt/ca.key 2048

使用這組私鑰來產生自簽證書 ca.crt

openssl req -new -x509 -days 365 -sha256 \
    -subj "/[C=TW/ST=Taipei/O=FOO/OU=BAR/CN=ken00535.com/[email protected]](mailto:C=TW/ST=Taipei/O=FOO/OU=BAR/CN=ken00535.com/[email protected])" \
    -key configs/mqtt/ca.key \
    -out configs/mqtt/ca.crt

再產生一組 server 用的私鑰 server.key

openssl genrsa -out configs/mqtt/server.key 2048

再產生憑證簽發的請求檔 csr

openssl req -new -sha256 \
    -key configs/mqtt/server.key \
    -subj "/C=TW/ST=Taipei/O=FOO/OU=BAR/CN=ken00535/[email protected]" 
    -out configs/mqtt/server.csr

使用 CA 簽發請求檔

openssl x509 -req -CAcreateserial -days 365 
    -CA configs/mqtt/ca.crt \
    -CAkey configs/mqtt/ca.key \
    -in configs/mqtt/server.csr \
    -out configs/mqtt/server.crt

完成一組球員兼裁判的簽發程序。簽發完成的證書 server.crt 可以用來做為 Broker 的證書,同樣的步驟跑一次,可以再產生一組 client.crt 做為 client 的證書,供雙向驗證用。為了方便,先都統一使用 server.crt。

接著產生一份 mosquitto 的設定檔,在其中加入

cafile configs/mqtt/ca.crt

# Path to the PEM encoded server certificate.
certfile configs/mqtt/server.crt

# Path to the PEM encoded keyfile.
keyfile configs/mqtt/server.key

放入認證用的證書。啟動 Broker 時帶入這組設定

mosquitto -c configs/mqtt/mosquitto.conf

Broker 就能被認證與加密了。

同理,Client 端也需要設定加密程序,新增 pkg/mqtt/tls.go 來產生加密設定

package mqtt

import (
    "crypto/tls"
    "crypto/x509"
    "io/ioutil"
)

func NewTLSConfig() *tls.Config {
    certpool := x509.NewCertPool()
    pemCerts, err := ioutil.ReadFile("configs/mqtt/ca.crt")
    if err == nil {
        certpool.AppendCertsFromPEM(pemCerts)
    }

    // Import client certificate/key pair
    cert, err := tls.LoadX509KeyPair("configs/mqtt/server.crt", "configs/mqtt/server.key")
    if err != nil {
        panic(err)
    }

    // Create tls.Config with desired tls properties
    return &tls.Config{
        RootCAs:            certpool,
        ClientAuth:         tls.NoClientCert,
        ClientCAs:          nil,
        InsecureSkipVerify: true,
        Certificates:       []tls.Certificate{cert},
    }
}

使用 tls.Config,在 RootCAs 放入 CA 證書;在 Certificates 放入客戶端的證書與私鑰。即可完成設定。

再修改 pkg/mqtt/mqtt.go,套用設定到 Client 端

tlsConfig := NewTLSConfig()
opts.SetTLSConfig(tlsConfig)

也要記得修改 tcp 為 tcps,表示套用 tls

opts := mqtt.NewClientOptions().AddBroker("tcps://127.0.0.1:1883").SetClientID(clientID)

MQTT.fx 也跟著加入設定

可以用 Wireshark 來比較一下加入 tls 前後的差別,在未加密的模式下,能輕易看到 Topic 跟 Payload 的訊息

如果是已加密,則只能看到 TLS

小結

MQTT 在 IoT 的應用上,應該會越來越常見,Golang 在 IoT 的優勢是跨平台編譯,只需要在 PC 上開發完後,就能編譯出 Arm 架構的 Binary Code,至於底層編譯有沒有對 Arm 優化就不得而知了。我覺得 Golang 的主要應用應該還是會落在 Server 端的 Application 開發,而 MQTT 比較像是扮演一個輕量的資料交換格式,用於初期或低負載的情境。

另個可以探索的題目是 mosquitto 的負載能力。因為 IoT 多是高頻輕負載的資訊傳輸,當連線數達到極限時,仍然要處理負載平衡的問題;應用端也可能需要有個 Message Queue 的緩衝機制,避免瞬間流量進來衝垮伺服器,或者沒有收到該收的資料。整個商業情境要處理的事還挺多的,也許後面有機會可以再試著討論看看。

Reference:

Read more

Weekly Issue 第 2 期:Linux 基金會啟動 FAIR 專案

有些產品看到會覺得行不通,有些產品則相反,只要聽到就覺得是個好主意。Sentry 的產品通常都是後者。我猜有部分,也是因為它們的產品都指向同一個使命:可除錯性。 🗞️ 熱門新聞 Linux Foundation Announces the FAIR Package Manager Project for Open Source Content Management System Stability Linux 基金會啟動 FAIR 專案,為 WordPress 外掛程式提供替代方案。 底下的 Supporting Quotes 可以看看,講話都很客氣,左一句「去中心化」右一句「透明的治理架構」,在講什麼大家都很清楚 😜 。 Uber 與 Airbnb 重塑 VC 玩法,一文看懂 a16z 創辦人

By Ken Chen

Weekly Issue 第 1 期:Stack Overflow 流量大跌

來自阮一峰老師的靈感與嘗試,我會在 Weekly Issue 中記錄每周值得分享的科技內容,周一發刊。多數內容都有刊在我的 X、Threads 或 Facebook 中,你可以追蹤上述社群媒體得到最新消息。這裡的性質更接近單周回顧與歷史歸檔。 🗞️ 熱門新聞 The Pulse #134: Stack overflow is almost dead StackOverflow 的情況比我想的還糟,退化到剛成立三個月的狀況?太要命了。 我自己好奇的是 2020 年的衰退如何引起?平台治理的問題嗎? Cloudflare service outage June 12, 2025 Cloudflare 近幾次中斷事故都有出報告,內容包括背景跟時間軸,還有改善方式,這是很正確也很重要的實踐。 我也曾經遇過幾次重要的服務停機事件,當時都會盡可能擠出時間即時更新 + 出報告。後來服務也的確越來越穩。這種問題很多都是文化層面的問題。 Ask HN: How

By Ken Chen
自訂網域很難嗎?DNS 的限制與實踐

自訂網域很難嗎?DNS 的限制與實踐

自訂網域(Custom Domain)是 SaaS 常見的服務,只是我通常都沒花錢買。某次跟朋友聊天,她想聽聽我對內容平台的觀點,嘰哩呱啦分析完一堆後,我最後建議她,最好還是買個網域: 「你想想看,妳現在投入這麼多心力在經營內容,建立自己的品牌形象。如果妳的網址永遠都掛在別人的平台底下,就像在別人家租房子,雖然方便,但終究不是自己的。」 「有了自訂網域,妳的品牌就是自己的,無論未來平台怎麼變,妳的讀者都能透過固定的網址找到妳,這對品牌來說很重要。就算未來妳想換平台,也不會流失妳辛苦建立起來的流量。」 後來我在 Ghost 官方頁面看到類似說法 If you would like to make your site memorable and easy to find with a branded custom domain, then you can

By Ken Chen