Centos7搭建Mosquitto，Golang使用MQTT

介紹

MQTT是一種機(jī)器到機(jī)器消息協(xié)議，旨在為“物聯(lián)網(wǎng)”設(shè)備提供輕量級(jí)發(fā)布/訂閱通信。它通常用于車輛的地理跟蹤車隊(duì)，家庭自動(dòng)化，環(huán)境傳感器網(wǎng)絡(luò)和公用事業(yè)規(guī)模的數(shù)據(jù)收集。
Mosquitto是一個(gè)實(shí)現(xiàn)了MQTT3.1協(xié)議的MQTT服務(wù)器（或代理 ，在MQTT中的用法），具有良好的社區(qū)支持，易于安裝和配置。
在我的應(yīng)用場(chǎng)景中，云服務(wù)器和食堂消費(fèi)終端需要數(shù)據(jù)交互，終端拉取服務(wù)器數(shù)據(jù)使用HTTP，服務(wù)器通知終端使用MQTT。 本文的主旨在于記錄Mosquitto服務(wù)的安裝和使用、基于Golang實(shí)現(xiàn)發(fā)布訂閱，以備日后查閱。

一、服務(wù)器安裝Mosquitto

服務(wù)器 CentOS Linux release 7.3.1611 (Core)
檢查系統(tǒng)是否已安裝EPEL(Extra Packages for Enterprise Linux)軟件源。

檢查安裝epel源

檢查 epel 是否已安裝

// 如下已存在epel源，則不需要安裝
? yum repolist | grep epel
!epel/x86_64               Extra Packages for Enterprise Linux 7 - x86_64 13,242

如果不存在，則需要安裝 epel

? yum -y install epel-release

安裝mosquitto

? yum -y install mosquitto

啟動(dòng)mosquitto

// 啟動(dòng)
? systemctl start mosquitto
// 停止
? systemctl stop mosquitto
// 重新啟動(dòng)
? systemctl restart mosquitto
// 查看運(yùn)行狀態(tài) 
? systemctl status mosquitto
// 設(shè)為開機(jī)啟動(dòng)
? systemctl enable mosquitto

測(cè)試Mosquitto

開啟兩個(gè)終端窗口，一個(gè)負(fù)責(zé)訂閱，另一個(gè)負(fù)責(zé)發(fā)布。


// 訂閱
? mosquitto_sub -h localhost -t test

// 發(fā)布
? mosquitto_pub -h localhost -t test -m "hello world"
  • -h host, 服務(wù)器的地址
  • -t topic，發(fā)布的主題，訂閱的主題和發(fā)布的一致才會(huì)收到信息
  • -m message, 消息內(nèi)容

mosquitto_pub 和 mosquitto_sub 參數(shù)詳解可以使用 man命令查看，或訪問官方地址： mosquitto_pub命令詳解 mosquitto_sub命令詳解

修改Mosquitto配置

配置文件位置/etc/mosquitto/mosquitto.conf，默認(rèn)無密碼可以訪問，端口號(hào)為1883。設(shè)置一下用戶名和密碼，mqtt 才會(huì)比較安全。
原配置文件內(nèi)容過多，為方便查看，新建一個(gè)空白文件代替默認(rèn)的。

  • 新建用戶名密碼
// uname1 為你的用戶名
? mosquitto_passwd -c /etc/mosquitto/passwd username1

// 如果要添加多個(gè)用戶使用 -b 參數(shù) 
// 必須在控制臺(tái)輸入明文的密碼，且文件不會(huì)覆蓋之前的
? mosquitto_passwd -b /etc/mosquitto/passwd username2 password2
  • 備份默認(rèn)配置
? mv /etc/mosquitto/mosquitto.conf /etc/mosquitto/mosquitto.conf.example
  • 新建文件 vim /etc/mosquitto/mosquitto.conf，寫入如下內(nèi)容
# 禁止匿名訪問
allow_anonymous false
# 用戶及密碼存儲(chǔ)文件
password_file /etc/mosquitto/passwd
# 端口號(hào)
port 8810

重啟測(cè)試

  • 重啟mosquitto
? systemctl restart mosquitto
  • 不帶用戶名和密碼進(jìn)行測(cè)試
// 指定端口號(hào)8810
// 不帶用戶名和密碼，訪問被拒絕
? mosquitto_pub -h localhost -t test -p 8810 -m "hello world"
Connection error: Connection Refused: not authorised.
  • 使用用戶名和密碼測(cè)試發(fā)布訂閱

在一個(gè)窗口執(zhí)行訂閱：

// -u 指定用戶名
// -P 指定密碼
? mosquitto_sub -h localhost -t test -p 8810  -u 'username1' -P 'password1'

在另一個(gè)終端窗口發(fā)布消息：

? mosquitto_pub -h localhost -t test -p 8810  -u 'username1' -P 'password1' -m 'hello world'

防火墻開放端口

服務(wù)器使用的firewall，如果需要遠(yuǎn)程測(cè)試，需要打開mosquitto端口

// 查看所有打開的端口
? firewall-cmd --zone=public --list-ports
// 打開8810端口
? firewall-cmd --zone=public --add-port=8810/tcp --permanent
// 重啟防火墻
? firewall-cmd --reload

二、Golang發(fā)布訂閱MQTT

項(xiàng)目目錄結(jié)構(gòu)：

|____mqtt
| |____mqtt.go
|____mqtt_pub.go
|____mqtt_sub.go
|____service
| |____lib.go

封裝mqtt

使用包 github.com/eclipse/paho.mqtt.golang 封裝，修改其中的配置 Host、UserName、Password。
新建文件mqtt/mqtt.go：

package mqtt

import (
    "encoding/json"
    "errors"
    "fmt"
    gomqtt "github.com/eclipse/paho.mqtt.golang"
    "strings"
    "sync"
    "time"
)

// mqtt服務(wù)器配置
const (
    Host     = "127.0.0.1:8810"
    UserName = "test"
    Password = "123456"
)

type Client struct {
    nativeClient  gomqtt.Client
    clientOptions *gomqtt.ClientOptions
    locker        *sync.Mutex
    // 消息收到之后處理函數(shù)
    observer func(c *Client, msg *Message)
}

type Message struct {
    // client_id
    ClientID string `json:"client_id"`
    // 接口名，訂閱號(hào)通過識(shí)別接口名處理相應(yīng)業(yè)務(wù)
    Action string `json:"action"`
    // 數(shù)據(jù)類型
    Type string `json:"type"`
    // 發(fā)布時(shí)間
    Time int64 `json:"time"`
    // 業(yè)務(wù)數(shù)據(jù)的header，可以攜帶一些系統(tǒng)參數(shù)
    Header interface{} `json:"header"`
    // 業(yè)務(wù)數(shù)據(jù)的body，業(yè)務(wù)參數(shù)
    Body interface{} `json:"body"`
}

func NewClient(clientId string) *Client {
    clientOptions := gomqtt.NewClientOptions().
        AddBroker(Host).
        SetUsername(UserName).
        SetPassword(Password).
        SetClientID(clientId).
        SetCleanSession(false).
        SetAutoReconnect(true).
        SetKeepAlive(120 * time.Second).
        SetPingTimeout(10 * time.Second).
        SetWriteTimeout(10 * time.Second).
        SetOnConnectHandler(func(client gomqtt.Client) {
            // 連接被建立后的回調(diào)函數(shù)
            fmt.Println("Mqtt is connected!", "clientId", clientId)
        }).
        SetConnectionLostHandler(func(client gomqtt.Client, err error) {
            // 連接被關(guān)閉后的回調(diào)函數(shù)
            fmt.Println("Mqtt is disconnected!", "clientId", clientId, "reason", err.Error())
        })

    nativeClient := gomqtt.NewClient(clientOptions)

    return &Client{
        nativeClient:  nativeClient,
        clientOptions: clientOptions,
        locker:        &sync.Mutex{},
    }
}

func (client *Client) GetClientID() string {
    return client.clientOptions.ClientID
}

func (client *Client) Connect() error {
    return client.ensureConnected()
}

// 確保連接
func (client *Client) ensureConnected() error {
    if !client.nativeClient.IsConnected() {
        client.locker.Lock()
        defer client.locker.Unlock()
        if !client.nativeClient.IsConnected() {
            if token := client.nativeClient.Connect(); token.Wait() && token.Error() != nil {
                return token.Error()
            }
        }
    }
    return nil
}

// 發(fā)布消息
// retained: 是否保留信息
func (client *Client) Publish(topic string, qos byte, retained bool, data []byte) error {
    if err := client.ensureConnected(); err != nil {
        return err
    }

    token := client.nativeClient.Publish(topic, qos, retained, data)
    if err := token.Error(); err != nil {
        return err
    }

    // return false is the timeout occurred
    if !token.WaitTimeout(time.Second * 10) {
        return errors.New("mqtt publish wait timeout")
    }

    return nil
}

// 消費(fèi)消息
func (client *Client) Subscribe(observer func(c *Client, msg *Message), qos byte, topics ...string) error {
    if len(topics) == 0 {
        return errors.New("the topic is empty")
    }

    if observer == nil {
        return errors.New("the observer func is nil")
    }

    if client.observer != nil {
        return errors.New("an existing observer subscribed on this client, you must unsubscribe it before you subscribe a new observer")
    }
    client.observer = observer

    filters := make(map[string]byte)
    for _, topic := range topics {
        filters[topic] = qos
    }
    client.nativeClient.SubscribeMultiple(filters, client.messageHandler)

    return nil
}

func (client *Client) messageHandler(c gomqtt.Client, msg gomqtt.Message) {
    if client.observer == nil {
        fmt.Println("not subscribe message observer")
        return
    }
    message, err := decodeMessage(msg.Payload())
    if err != nil {
        fmt.Println("failed to decode message")
        return
    }
    client.observer(client, message)
}

func decodeMessage(payload []byte) (*Message, error) {
    message := new(Message)
    decoder := json.NewDecoder(strings.NewReader(string(payload)))
    decoder.UseNumber()
    if err := decoder.Decode(&message); err != nil {
        return nil, err
    }
    return message, nil
}

func (client *Client) Unsubscribe(topics ...string) {
    client.observer = nil
    client.nativeClient.Unsubscribe(topics...)
}

map解析成struct

基于包github.com/goinggo/mapstructure 將訂閱者收到的map解析為相應(yīng)的struct。
新建文件 service/lib.go：

package service

import "github.com/goinggo/mapstructure"

// User 一個(gè)用于測(cè)試的struct
type User struct {
    ID   int    `json:"id"`
    Name string `json:"name"`
    Age  int    `json:"age"`
}

// MapToStruct map轉(zhuǎn)struct.
// mqtt解析字段直接使用項(xiàng)目通用的json標(biāo)簽
// mqtt傳輸int會(huì)轉(zhuǎn)為string(int)，需要開啟 WeaklyTypedInput
func MapToStruct(m interface{}, structPointer interface{}) error {
    // https://godoc.org/github.com/mitchellh/mapstructure#DecoderConfig
    config := &mapstructure.DecoderConfig{
        TagName:          "json",
        Result:           structPointer,
        WeaklyTypedInput: true,
    }
    newDecode, _ := mapstructure.NewDecoder(config)

    if err := newDecode.Decode(m); err != nil {
        return err
    }
    return nil
}

發(fā)布者

新建文件 mqtt_pub.go：

package main

import (
    "encoding/json"
    "github.com/astaxie/beego/logs"
    "github.com/gushasha/go-mqtt/mqtt"
    "github.com/gushasha/go-mqtt/service"
    "time"
)

func main() {

    const (
        clientId = "pub-001"
        // topic規(guī)則：設(shè)備編號(hào)/接口名
        topicName      = "device001/user"
        actionNameUser = "user/detail"
    )

    client := mqtt.NewClient(clientId)
    err := client.Connect()
    if err != nil {
        logs.Error(err.Error())
    }

    // 發(fā)布一個(gè) user 消息
    body := service.User{1, "小寶", 2}

    msg := &mqtt.Message{
        ClientID: clientId,
        Action:   actionNameUser,
        Type:     "json",
        Time:     time.Now().Unix(),
        Body:     body,
    }
    data, _ := json.Marshal(msg)
    err = client.Publish(topicName, 0, false, data)
    if err != nil {
        panic(err)
    }
}

訂閱者

新建文件 mqtt_sub.go：

package main

import (
    "fmt"
    "github.com/gushasha/go-mqtt/mqtt"
    "github.com/gushasha/go-mqtt/service"
    "sync"
)

var (
    wg sync.WaitGroup
)

const (
    clientId = "client-001"
    topicName      = "device001/#"
    actionNameUser = "user/detail"
)

func main() {

    client := mqtt.NewClient(clientId)
    err := client.Connect()
    if err != nil {
        panic(err.Error())
    }

    wg.Add(1)
    err = client.Subscribe(userHandler, 0, topicName)
    if err != nil {
        panic(err)
    }
    wg.Wait()
}

var userHandler = func(c *mqtt.Client, msg *mqtt.Message) {
    switch msg.Action {
    case actionNameUser:
        // map 轉(zhuǎn) service.User
        user := &service.User{}
        err := service.MapToStruct(msg.Body, user);
        if err != nil {
            fmt.Println("error:", err)
        } else {
            HandleUser(user)
        }
    default:
        fmt.Printf("unkonwn action %s \n", msg.Action)
    }
}

func HandleUser(user *service.User) {
    fmt.Printf("user: %#v \n", user)
}

執(zhí)行測(cè)試

// 在一個(gè)終端窗口執(zhí)行訂閱
? go run mqtt_sub.go

// 另一個(gè)終端窗口執(zhí)行發(fā)布
? go run mqtt_pub.go

// 訂閱端收到信息，解析為struct：
// mqttuser: &service.User{ID:1, Name:"小寶", Age:2}

參考資料

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

友情鏈接更多精彩內(nèi)容