【物联网IoT - 自部署MQTT Broker服务】

常见的MQTT Broker

• Eclipse Mosquitto：轻量级 MQTT Broker，易于部署和使用。
• EMQX：高性能、可扩展的 MQTT Broker。
• HiveMQ：企业级 MQTT Broker，支持集群和高可用性。
• VerneMQ：高性能、可扩展的 MQTT Broker，支持集群。

需求

• 配置要求低：单机可部署
• 适配mqtt协议，客户端支持paho.mqtt
• 部署相对简单
• 部署环境支持

选哪个

选Eclipse Mosquitto

为什么

1. 之前的客户端用了paho mqtt （也属于Eclipse下面的项目），和Eclipse Mosquitto 都属于Eclipse 基金会下的开源项目，完全适配。
2. 部署简单，且支持多环境。

Eclipse Mosquitto 作为物联网IoT MQTT服务，对比其它MQTT Broker，类似于数据库中的SQLite。

单机可部署，轻量级，不支持分布式，简单快直接。

部署步骤

1.安装

sudo apt-get update

安装客户端和服务端

sudo apt-get install -y mosquitto mosquitto-clients

2. 启动

mosquitto -c /etc/mosquitto/mosquitto.conf

3. 测试

使用mosquitto命令行客户端 mosquitto_sub/pub 安装完后试下，订阅和发布

mosquitto_sub -h localhost -t test/topic

mosquitto_pub -h localhost -t test/topic -m "Hello, MQTT"

代码 测试，一个Go Paho MQTT Client示例

package main

import (
    "fmt"
    "time"
    mqtt "github.com/eclipse/paho.mqtt.golang"
)

func main() {
    // 创建 MQTT 客户端选项
    opts := mqtt.NewClientOptions().AddBroker("tcp://123.56.56.123:1883")
    opts.SetClientID("go_mqtt_client")
    // opts.SetUsername("username") // 添加用户名
    // opts.SetPassword("xxx")   // 添加密码
    opts.SetDefaultPublishHandler(messagePubHandler)

    // 创建 MQTT 客户端
    client := mqtt.NewClient(opts)
    if token := client.Connect(); token.Wait() && token.Error() != nil {
        panic(token.Error())
    }

    // 订阅主题
    topic := "test/topic"
    token := client.Subscribe(topic, 1, nil)
    token.Wait()
    fmt.Printf("Subscribed to topic: %s\n", topic)

    // 发布消息
    for i := 0; i < 5; i++ {
        text := fmt.Sprintf("Hello, MQTT %d", i)
        token = client.Publish(topic, 0, false, text)
        token.Wait()
        time.Sleep(1 * time.Second)
    }

    // 断开连接
    client.Disconnect(250)
}

// 消息发布处理函数
var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
    fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())
}

4 安全相关

第一步：开启ssl加密传输，配置ssl，避免消息在网络中明文传输。

1. 使用RSA算法生成，私钥，公钥，证书信息。

openssl genpkey -algorithm RSA -out mosquitto.key
openssl req -new -key mosquitto.key -out mosquitto.csr
openssl x509 -req -in mosquitto.csr -signkey mosquitto.key -out mosquitto.crt -days 365

2. 在mosquitto中配置

listener 8883
cafile /path/to/ca.crt
certfile /path/to/mosquitto.crt
keyfile /path/to/mosquitto.key

3. 重启服务后，再次连接，注意代码中加上ssl配置

// 更新连接串
opts := mqtt.NewClientOptions().AddBroker("ssl://123.56.56.123:8883")
...
// 配置TLS
tlsConfig := &tls.Config{
    InsecureSkipVerify: true, // 跳过对服务端CA证书的验证
}
opts.SetTLSConfig(tlsConfig)

第二步：身份验证，增加密码

mosquitto_passwd -c /etc/mosquitto/passwd username

修改配置文件 ，路径 /etc/mosquitto/mosquitto.conf

allow_anonymous false
password_file /etc/mosquitto/passwd

对应代码修改

...
// 创建 MQTT 客户端选项
opts := mqtt.NewClientOptions().AddBroker("tcp://123.56.56.123:1883")
opts.SetClientID("go_mqtt_client")
opts.SetUsername("username") // 添加用户名
opts.SetPassword("xxx")   // 添加密码
opts.SetDefaultPublishHandler(messagePubHandler)
...

补充问题

mosquitto对于传输大小是否有限制

• 默认限制 256MB ，可以设置

max_packet_size 268435455  # 设置最大消息大小为256MB

mosquitto里面的消息队列长度默认多长，如果长时间不消费，怎么处理死信

• 有过期时间， 比如1小时

message-expiry-interval 3600  # 消息过期时间为1小时