Go MQTT 实战：从设备数据上报到自动重连与可靠消费

设备每隔几秒上报一次温度，服务端接收并记录；设备断网后，后台还能看到它已经离线。这类需求很适合用 MQTT 串起来。

真正接入项目时，需要处理的不只是 Publish 和 Subscribe：连接成功是否代表订阅已就绪？断线重连后谁来恢复订阅？发布超时还能不能重试？QoS 1 是否意味着数据库里只会出现一条记录？

这篇文章用一个可以复制运行的 Go 示例，把本地 Broker、温度上报、订阅处理、自动重连、设备状态与正常退出串成完整流程，再说明持久化、业务幂等和生产配置应该补在哪一层。

一、先明确这次要实现的链路

我们用两个 Go 进程分别模拟设备和服务端：

设备进程（pub） ──发布温度──> Mosquitto Broker ──转发──> 订阅进程（sub）
                                │
                                └──保存每个状态 Topic 的最新 retained 消息

发布者不需要知道订阅者的地址，双方通过 Broker 和 Topic 联系。多个普通订阅者订阅同一主题时，各自都会收到消息，不能直接把它当成“多个 worker 竞争一条任务”的队列。

本文采用 MQTT 3.1.1，客户端固定为 github.com/eclipse/paho.mqtt.golang v1.5.1，方便复现。这个模块的 go.mod 要求 Go 1.24.0 或更高；开发和部署应使用仍受维护的 Go 工具链。这里的固定版本是示例基线，不代表最新版本。Paho 版本依赖声明

如果项目需要 MQTT 5 的消息过期时间、原因码等特性，应评估另一套客户端 github.com/eclipse/paho.golang；它和本文模块的 API 不能直接混用。Paho 客户端说明

Topic 和消息结构

先约定两类主题：

Topic	含义	发布 QoS	Retain
lab/devices/sensor-01/telemetry	温度采样事件	1	false
lab/devices/sensor-01/status	当前在线状态	1	true

温度消息使用 JSON：

{
  "event_id": "c389ff02fa136a9e",
  "device_id": "sensor-01",
  "temperature": 23.5,
  "time": "2026-11-03T01:00:00Z"
}

event_id 标识一次业务事件，后面用来做幂等；device_id 标识设备，不能用 MQTT 协议里会复用的 Packet ID 替代。

订阅过滤器 lab/devices/# 会匹配设备的温度和状态；如果只关心温度，可以改成 lab/devices/+/telemetry。+ 匹配一层，# 匹配多层且必须位于过滤器末尾；发布 Topic 不能带这两个通配符。Topic 区分大小写。MQTT 3.1.1 主题规则

二、用 Docker 启动本地 Mosquitto

新建工作目录和配置：

mkdir go-mqtt-demo
cd go-mqtt-demo
mkdir -p mosquitto/config
cat > mosquitto/config/mosquitto.conf <<'EOF'
listener 1883
allow_anonymous true
persistence false
log_dest stdout
EOF

创建 compose.yaml：

services:
  mqtt:
    image: eclipse-mosquitto:2
    ports:
      - "127.0.0.1:1883:1883"
    volumes:
      - ./mosquitto/config/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro

运行：

docker compose up -d
docker compose logs mqtt

这里显式开放容器内的监听，并把宿主机端口绑定在 127.0.0.1，供本机实验使用。匿名访问和 persistence false 是本地演示配置：Broker 重启后，内存里的 retained 状态和会话数据不会被这份配置保留下来。上线前需要配置认证、权限和持久化；eclipse-mosquitto:2 也是会更新的主版本标签，生产环境应锁定经过验证的补丁版本或镜像摘要。Mosquitto 监听与持久化配置

三、完整 Go 代码：一个程序，两种模式

在刚才的目录初始化模块：

go mod init example.com/mqtt-demo
go get github.com/eclipse/paho.mqtt.golang@v1.5.1

创建下面的 main.go。程序默认订阅，也可以通过 -mode pub 模拟设备，每两秒上报一次固定温度。

main.go

package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

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

const (
	deviceID       = "sensor-01"
	telemetryTopic = "lab/devices/" + deviceID + "/telemetry"
	statusTopic    = "lab/devices/" + deviceID + "/status"
	topicFilter    = "lab/devices/#"
)

type Telemetry struct {
	EventID     string  `json:"event_id"`
	DeviceID    string  `json:"device_id"`
	Temperature float64 `json:"temperature"`
	Time        string  `json:"time"`
}

func randomID() (string, error) {
	var b [8]byte
	if _, err := rand.Read(b[:]); err != nil {
		return "", err
	}
	return hex.EncodeToString(b[:]), nil
}

func waitToken(t mqtt.Token, operation string) error {
	if !t.WaitTimeout(5 * time.Second) {
		return fmt.Errorf("%s: 等待超时，最终结果未知", operation)
	}
	if err := t.Error(); err != nil {
		return fmt.Errorf("%s: %w", operation, err)
	}
	return nil
}

func onMessage(_ mqtt.Client, m mqtt.Message) {
	if !strings.HasSuffix(m.Topic(), "/telemetry") {
		log.Printf("状态 topic=%s retained=%t payload=%s", m.Topic(), m.Retained(), m.Payload())
		return
	}
	var event Telemetry
	if err := json.Unmarshal(m.Payload(), &event); err != nil {
		log.Printf("无效 JSON topic=%s: %v", m.Topic(), err)
		return
	}
	if event.EventID == "" || event.DeviceID == "" {
		log.Printf("缺少业务标识 topic=%s", m.Topic())
		return
	}
	log.Printf("收到 event_id=%s device=%s temperature=%.1f time=%s",
		event.EventID, event.DeviceID, event.Temperature, event.Time)
}

func run() error {
	mode := flag.String("mode", "sub", "sub 或 pub")
	broker := flag.String("broker", "tcp://127.0.0.1:1883", "MQTT broker 地址")
	id := flag.String("id", "", "唯一的 MQTT Client ID，默认随机生成")
	flag.Parse()
	if *mode != "pub" && *mode != "sub" {
		return fmt.Errorf("mode 必须为 pub 或 sub")
	}
	if *id == "" {
		suffix, err := randomID()
		if err != nil {
			return err
		}
		*id = "go-" + *mode + "-" + suffix
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	opts := mqtt.NewClientOptions().AddBroker(*broker).SetClientID(*id)
	opts.SetProtocolVersion(4) // MQTT 3.1.1 的协议级别是 4。
	opts.SetCleanSession(true)
	opts.SetAutoReconnect(true)
	opts.SetConnectRetry(false) // 首次连接失败直接返回；断线后自动重连。
	opts.SetConnectTimeout(5 * time.Second)
	opts.SetWriteTimeout(5 * time.Second)
	opts.SetKeepAlive(20 * time.Second)
	opts.SetPingTimeout(5 * time.Second)
	opts.SetMaxReconnectInterval(10 * time.Second)
	opts.SetDefaultPublishHandler(onMessage)
	if user := os.Getenv("MQTT_USERNAME"); user != "" {
		opts.SetUsername(user)
		opts.SetPassword(os.Getenv("MQTT_PASSWORD"))
	}
	if *mode == "pub" {
		opts.SetWill(statusTopic, `{"online":false}`, 1, true)
	}
	opts.SetConnectionLostHandler(func(_ mqtt.Client, err error) {
		log.Printf("连接断开，将尝试重连: %v", err)
	})
	ready := make(chan error, 1)
	opts.SetOnConnectHandler(func(c mqtt.Client) {
		var err error
		if *mode == "sub" {
			t := c.Subscribe(topicFilter, 1, onMessage)
			err = waitToken(t, "订阅")
			if err == nil {
				// SUBACK 可能返回订阅拒绝，不能只检查 Token.Error。
				if st, ok := t.(*mqtt.SubscribeToken); ok {
					qos, exists := st.Result()[topicFilter]
					if !exists || qos > 2 {
						err = fmt.Errorf("订阅未获授权: %s", topicFilter)
					}
				}
			}
		} else {
			err = waitToken(c.Publish(statusTopic, 1, true, `{"online":true}`), "发布在线状态")
		}
		if err != nil {
			log.Printf("连接初始化失败: %v", err)
		} else {
			log.Printf("已就绪 mode=%s client_id=%s", *mode, *id)
		}
		select {
		case ready <- err:
		default:
		}
	})

	client := mqtt.NewClient(opts)
	defer client.Disconnect(1000) // 单位是毫秒；业务停机逻辑在下面执行。
	if err := waitToken(client.Connect(), "连接"); err != nil {
		return err
	}
	select {
	case err := <-ready:
		if err != nil {
			return err
		}
	case <-ctx.Done():
		return nil
	case <-time.After(12 * time.Second):
		return fmt.Errorf("等待连接初始化超时")
	}
	if *mode == "sub" {
		<-ctx.Done()
		return nil
	}

	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			if client.IsConnectionOpen() {
				if err := waitToken(client.Publish(statusTopic, 1, true, `{"online":false}`), "发布离线状态"); err != nil {
					log.Printf("离线状态未确认: %v", err)
				}
			}
			return nil
		case now := <-ticker.C:
			if !client.IsConnectionOpen() {
				log.Print("尚未恢复连接，跳过本次采样")
				continue
			}
			eventID, err := randomID()
			if err != nil {
				return err
			}
			event := Telemetry{eventID, deviceID, 23.5, now.UTC().Format(time.RFC3339Nano)}
			payload, err := json.Marshal(event)
			if err != nil {
				return err
			}
			if err := waitToken(client.Publish(telemetryTopic, 1, false, payload), "发布温度"); err != nil {
				log.Printf("event_id=%s: %v", eventID, err)
				continue
			}
			log.Printf("发布流程完成 event_id=%s", eventID)
		}
	}
}

func main() {
	if err := run(); err != nil {
		log.Print(err)
		os.Exit(1)
	}
}

然后整理依赖并编译：

go mod tidy
go build -o mqtt-demo .

示例保留了 JSON 编码错误、连接错误、订阅返回码和发布等待结果的检查。main 在 run 返回后退出，保证 run 内的 defer 有机会执行。

四、启动订阅和发布，观察完整过程

在第一个终端启动订阅者：

./mqtt-demo -mode sub -id backend-01

等日志出现 已就绪 mode=sub，再在第二个终端启动设备：

./mqtt-demo -mode pub -id device-01

订阅端会看到类似输出，随机 ID 和时间以实际运行为准：

已就绪 mode=sub client_id=backend-01
状态 topic=lab/devices/sensor-01/status retained=false payload={"online":true}
收到 event_id=c389ff02fa136a9e device=sensor-01 temperature=23.5 time=2026-11-03T01:00:00Z

当前订阅者收到实时状态时，retained 可以是 false；如果新开一个订阅进程，Broker 会把保存的最新状态发给它，这时会看到 retained=true。温度事件没有设置 Retain，新订阅者不会因此获得全部历史采样。MQTT 保留消息语义

本例的 deviceID 固定为 sensor-01，同一时间只运行一个设备发布进程。-id 只改变连接身份，不会改变业务设备和 Topic；扩展多设备时，应同时把设备 ID 和 Topic 参数化。

试验一：正常退出

在发布进程所在终端按 Ctrl+C。程序停止采样，尝试发布 retained 的 online=false，随后调用 Disconnect(1000) 等待客户端退出。

这个等待参数的单位是毫秒。它只约束客户端的退出等待，不代表订阅端已经处理完全部消息，也不能替代数据库事务或业务任务排空。

试验二：异常退出与遗嘱

重新启动发布者，再从另一个终端查找该 mqtt-demo 进程的 PID，发送 kill -9 <PID>。Broker 感知连接异常后，会发布连接建立时登记的遗嘱：{"online":false}。

遗嘱可能在 TCP 断开时较快触发，也可能需要等待 Keep Alive 检测到失联。正常发送 MQTT DISCONNECT 时，Broker 不会再发送该连接的遗嘱，因此正常退出路径要主动更新状态。遗嘱只反映 Broker 对连接的判断；健康检查还应结合最近上报时间。若正常退出前的离线发布失败，retained 状态也可能暂时不准确。MQTT 遗嘱与 Keep Alive

试验三：断线后恢复订阅

保持两个进程运行，执行：

docker compose restart mqtt

预期能观察到断线日志、重新就绪和新的温度消息。这是推荐读者执行的故障演练步骤；重连时间取决于 Broker 启动和网络情况。

这里订阅者使用 CleanSession=true，连接恢复后必须重新订阅。程序将 Subscribe 放在 OnConnect 中，每次连接成功都会执行。只在 main 中调用一次订阅，通常会导致“连接恢复了，但不再收到消息”。Paho 连接回调说明

五、代码里最容易漏掉的几个细节

1. 连接成功和订阅就绪是两件事

Connect 完成后，还需要发送 SUBSCRIBE 并等待 SUBACK。示例用 ready channel 等待第一次初始化结果，订阅成功后才输出就绪。

另外，SUBACK 可能携带拒绝码 0x80，所以不能只看 Token.Error()；代码还会检查 SubscribeToken.Result()。Broker 也可以授予低于申请值的 QoS，应以实际返回值和业务要求为准。Paho SubscribeToken

本例首次连接或订阅初始化失败会退出；建立过连接后的网络中断由客户端自动重连处理。重连后的订阅初始化失败目前只记录日志，没有额外的订阅重试循环。生产服务应把订阅状态纳入 readiness，并对临时失败做有界退避重试，不能仅凭 TCP 已连接就宣布服务可用。

2. 等待超时不等于操作被取消

Paho 的连接、发布、订阅返回 Token。WaitTimeout 返回 false 只表示调用方停止等待，Token 不会因此自动变成错误，底层操作也不一定停止。Paho Token 接口

因此，发布超时时应该记录“结果未知”，不能断言消息没有发出去。如果业务决定重试同一事件，必须复用原来的 event_id；重新生成 ID 会把同一业务动作伪装成另一条事件。

示例中的每次 tick 都代表一次新采样，失败仅记录，不自动补发。IsConnectionOpen 也只是调用瞬间的状态检查，不能保证随后的网络写入成功。Paho 在自动重连路径上的 Token 完成还有版本相关的处理细节，所以日志使用“发布流程完成”，不把它当成端到端业务确认。Paho 自动重连注意事项

3. Client ID 必须避免冲突

同一个 Broker 上同时连接的客户端应使用不同 Client ID。两个进程使用同一个 ID，会相互顶掉连接，表现为不断断线、重连。

示例默认生成随机 ID，适合临时实验。需要恢复持久会话的订阅者应使用稳定且唯一的 ID；部署多个副本时，要为每个副本分配身份，不能复制同一个固定值。

4. 回调里不要直接执行慢业务

示例回调只解析 JSON 并记录日志，方便观察。默认有序回调下，不要在处理函数里等待数据库慢查询、HTTP 请求，或同步等待另一次 MQTT 发布，否则可能阻塞后续消息和确认。Paho 回调阻塞说明

真实项目可将校验后的事件送入有界工作队列，再由固定 worker 处理。队列满了要明确选择背压、拒绝或落盘，并记录指标，不能无限启动 goroutine。

还要注意确认时机：默认自动 ACK 下，回调将消息放进内存队列后就返回，MQTT 确认可能早于业务落库。此时进程崩溃，队列里的消息仍可能丢失。若要求处理成功后才确认，需要结合 Paho 的 SetAutoAckDisabled(true)、Message.Ack() 以及 Broker 会话恢复策略设计，并验证所用 QoS 的确认流程；不能仅移动一行 ACK 就声称端到端可靠。Paho 手动确认配置

六、QoS 1 为什么仍然需要业务幂等

QoS 描述 MQTT 协议链路的投递行为：

级别	协议含义	常见取舍
QoS 0	至多一次，不做协议级重传确认	高频、允许偶发丢失的采样
QoS 1	至少一次，可能重复	有幂等处理的事件上报
QoS 2	协议层恰好一次，交互更多	需要该协议语义且接受额外开销的场景

发布者到 Broker、Broker 到订阅者是两个阶段。订阅者收到的 QoS 还受订阅授予值限制。QoS 2 也不等于数据库事务只提交一次，更不会自动让业务代码在崩溃后恢复。MQTT QoS 定义

例如 PostgreSQL 中可以建立事件唯一约束：

CREATE TABLE telemetry_events (
    event_id TEXT PRIMARY KEY,
    device_id TEXT NOT NULL,
    temperature DOUBLE PRECISION NOT NULL,
    sampled_at TIMESTAMPTZ NOT NULL
);

INSERT INTO telemetry_events (event_id, device_id, temperature, sampled_at)
VALUES ($1, $2, $3, $4)
ON CONFLICT (event_id) DO NOTHING;

这段 SQL 是生产接入思路，未集成到本文的日志示例中。若插入事件后还要更新设备状态，应把去重判断和状态更新放在同一事务里，只有新事件插入成功才执行后续变更。调用支付、发送通知等外部系统时，还需要对方支持幂等键，或采用 outbox 等可恢复的投递流程。

生产事件 ID 应使用有足够碰撞空间、作用域明确的标识方案。本文的随机 64 位值用于演示；大规模系统可以使用 UUID，或使用“设备 ID + 持久化序列号”，并明确设备重启后的序列恢复方式。

七、离线消息与 retained 消息各管什么

容易混淆的三个配置分别作用在不同位置：

机制	保存的是什么	需要注意什么
Retain	每个 Topic 的最新一条保留消息	适合当前状态，不是历史事件列表
持久会话	订阅关系及符合条件的待投递消息	依赖固定 Client ID、会话配置与 Broker 队列限制
客户端本地持久化	客户端侧协议状态或业务待发送事件	要明确磁盘写入、崩溃恢复和重试规则

本文 CleanSession=true，且 Broker 未开启磁盘持久化，所以不提供订阅者离线补发和进程崩溃后的事件恢复；设备断线期间的采样会被主动跳过。自动重连只帮助恢复连接。

如果需要订阅者断线后补收消息，至少要使用稳定 Client ID、CleanSession=false、合适的 QoS，并核对 Broker 的持久化、消息数和字节数上限。持久会话恢复时，消息可能在重新执行 Subscribe 前到达，因此要在连接前注册处理路由或默认回调。发布端要保证业务事件可恢复，还需要把待发送事件持久化；Paho 的协议存储不是完整的业务 outbox。Paho 会话与队列注意事项

清除一个 retained 状态，可以对相同 Topic 发布 retained 的零长度载荷：

docker compose exec mqtt mosquitto_pub   -h 127.0.0.1 -t lab/devices/sensor-01/status -q 1 -r -n

此时在线订阅者也可能看到空消息，业务处理应识别“状态被清除”，不要直接当作有效 JSON。

八、接入真实环境：认证、TLS 和权限

完整示例已经支持从 MQTT_USERNAME、MQTT_PASSWORD 读取凭据；Broker 端仍需配置对应用户和权限。密码通过明文 TCP 传输没有保密性，跨主机部署应使用 TLS。Mosquitto 认证说明

如果 Broker 使用受系统信任的服务端证书，在创建 client 之前为 opts 增加以下配置，并给 import 补上 crypto/tls：

opts.SetTLSConfig(&tls.Config{
    MinVersion: tls.VersionTLS12,
})

然后连接已配置 TLS 的 Broker：

export MQTT_USERNAME='device-user'
# 由运行环境或密钥管理系统注入 MQTT_PASSWORD。
./mqtt-demo -mode pub -id device-01 -broker ssl://mqtt.example.com:8883

这个命令需要真实可访问的 Broker；本地 Compose 仅提供 1883 明文监听。私有 CA 要加载进 RootCAs，双向 TLS 要配置客户端证书；连接主机名必须匹配证书。不要通过 InsecureSkipVerify 绕过证书校验。Paho TLS 配置接口

权限应按设备隔离：设备只能发布自己的 telemetry、status，服务端账号按需要订阅。服务端还应校验 Topic 中的设备身份与载荷一致，并限制消息大小、校验时间戳和数值范围，不能只信任 JSON 中的 device_id。

九、遇到问题时按这个顺序排查

现象	优先检查
connection refused	Broker 是否启动、端口映射和连接地址是否正确
Docker 容器里连不上 127.0.0.1	它指向当前容器；同一 Compose 网络应使用服务名 mqtt:1883
连接成功但收不到数据	是否等待订阅就绪、Topic 大小写、过滤器、ACL 和 SUBACK 返回码
不断断线、重连	Client ID 是否冲突，回调是否阻塞，Keep Alive 和代理超时是否匹配
重连后没有新消息	是否在 OnConnect 恢复订阅，恢复订阅是否失败
新订阅就收到旧状态	检查 retained 标志，这是保存的最新状态
数据库出现重复记录	是否按稳定事件 ID 做唯一约束和事务内幂等
发布等待超时	记录为结果未知，查 Broker 和网络日志，避免用新事件 ID 盲目重试

可以先绕过 Go 程序，用容器内的客户端观察消息：

docker compose exec mqtt mosquitto_sub   -h 127.0.0.1 -t 'lab/devices/#' -q 1 -v

若命令行能收到数据，再检查应用回调和业务处理；若命令行也收不到，先检查 Broker、认证和主题。停止实验环境可执行 docker compose down。

十、从演示程序走向业务服务

这份程序已经串起连接、订阅、上报、重连和状态通知，但它的消费结果只是日志。接入实际业务时，下一步应围绕具体要求补齐：订阅就绪状态、持久化事件、幂等事务、有界处理队列，以及断网、Broker 重启、进程崩溃后的恢复验证。

MQTT 帮助我们把设备和服务端通过主题连接起来；事件是否能恢复、是否会重复扣款、是否已经处理成功，仍需要业务系统给出明确答案。