Go Paho MQTT Go 怎样订阅多个不同 QoS 的主题

使用 SubscribeMultiple(filters map[string]byte, callback MessageHandler)，把每个主题过滤器作为map的键，把该过滤器请求的QoS作为值。一次调用即可表达不同QoS；调用后要等待Token完成，并读取每项授权结果，不能把请求值直接当成服务端最终批准值。

本文针对 github.com/eclipse/paho.mqtt.golang 的MQTT 3.1.1客户端，示例从“连接已经完成”的位置开始，不混用其他Paho包或MQTT 5客户端的接口。

官方地址：https://pkg.go.dev/github.com/eclipse/paho.mqtt.golang

把请求QoS放进过滤器映射

假设采集服务需要接收温度、告警与审计三个主题，分别请求QoS 0、1、2。SubscribeMultiple的map保留了每个过滤器自己的请求值，不必为了不同QoS建立三个客户端。本次调用使用同一个共享回调，回调可通过消息主题分派业务。

这只是展示映射方式的业务例子，不是工业设备必须采用的等级分配。实际请求应由消息用途、Broker能力和业务约束决定。

用一个函数处理等待与逐项结果

以下原创代码适合作为应用订阅模块。调用者传入已连接的client和handler；示例未在本文环境运行，也没有声称存在可用Broker。

package subscriptions

import (
    "fmt"
    "time"

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

// Install 接收已完成连接的客户端；连接及认证由调用方负责。
func Install(client mqtt.Client, handler mqtt.MessageHandler) (map[string]byte, error) {
    filters := map[string]byte{
        "plant/line-a/temperature": 0,
        "plant/line-a/alarm":       1,
        "plant/line-a/audit":       2,
    }

    token := client.SubscribeMultiple(filters, handler)
    // 超时仅表示本次等待结束，不表示后台操作已经取消。
    if !token.WaitTimeout(5 * time.Second) {
        return nil, fmt.Errorf("等待订阅回执超时，结果尚不确定")
    }
    if err := token.Error(); err != nil {
        return nil, fmt.Errorf("订阅操作失败: %w", err)
    }

    sub, ok := token.(*mqtt.SubscribeToken)
    if !ok {
        // 包装或模拟Client可能返回其他Token，显式报告无法读取逐项结果。
        return nil, fmt.Errorf("当前Token不提供SubscribeToken结果")
    }
    result := sub.Result()
    granted := make(map[string]byte, len(filters))
    for filter := range filters {
        qos, exists := result[filter]
        if !exists {
            return granted, fmt.Errorf("过滤器 %q 缺少回执", filter)
        }
        // MQTT 3.1.1中成功返回0、1、2；0x80表示订阅失败。
        if qos > 2 {
            return granted, fmt.Errorf("过滤器 %q 被拒绝，返回码0x%02x", filter, qos)
        }
        granted[filter] = qos
    }
    return granted, nil
}

返回的granted包含已处理的成功结果；遇到后续拒绝时，它可能只有部分条目。这个局部返回约定方便调用方记录已知状态，但调用方不能把它理解为其他主题已自动撤销。错误意味着应用需要决定如何处理整个订阅集合。

例如所有主题都是启动必需项时，可由上层把状态标记为“订阅未就绪”，再安排退订或断开；若允许部分运行，就明确列出哪些功能可用。不要在函数返回错误后仍发送笼统的“全部订阅成功”通知。

Token完成后还要逐项读取结果

Paho的SubscribeToken.Result返回按过滤器组织的Broker返回码。MQTT 3.1.1的SUBACK成功码可以是0、1、2，失败码为0x80，服务端允许授予低于请求的级别。因此申请QoS 2并不等于一定获得2，更不等于每条收到的消息都具有QoS 2。

按协议，向订阅者交付时的QoS受原消息QoS与授予订阅的最大QoS共同约束。应把配置中的请求值、SUBACK授权值和消息实际QoS分别看待。

另外，WaitTimeout返回false时，不能只看Error是否为nil就宣布成功。官方Token说明超时不会自动设置错误，并且操作可能仍在继续。示例因此先处理等待结果，再检查Error，最后读取Result。这三类判断不能互相替代。

把权限、重连与回调交给不同职责

部署前建议按过滤器核对账号订阅权限，而不是只确认“账号能连接”。连接通过与主题被授权属于不同判断。日志记录过滤器、请求QoS、授权返回码和错误类别即可，凭据与完整业务载荷不宜直接写入普通运行日志。

订阅配置建议作为只读输入维护，调用期间不要让其他goroutine修改同一map。需要动态变化时，先复制一份本次配置，再交给订阅函数；这样回执对应的是明确的一组请求。

断线恢复后是否重建订阅，应结合CleanSession、Broker会话和应用策略确定。建议把同一份过滤器配置集中放在一个安装订阅的入口，由连接生命周期调用，不要让消息回调、定时任务和重连处理同时各自订阅一遍。

共享回调应快速返回，并具备并发使用所需的同步措施。Paho文档特别提醒默认OrderMatters为true时，回调不能阻塞或在其中直接等待可能阻塞的客户端调用。耗时业务可交给有容量约束的工作队列，并明确队列满时如何处理，而不是无限制创建goroutine。

相关问题

同一个过滤器能在map里写两个QoS吗

不能。map的同一个键只有一个最终值。若是更新同一过滤器的订阅，应明确进行一次新的订阅请求，并读取新的授权结果。

每个主题需要不同处理函数怎么办

SubscribeMultiple接收一个回调。可在共享回调里分派，也可根据应用需要分别调用Subscribe传入各自回调；差别是应用组织方式，不是请求不同QoS必须建立多连接。

收到0x80后其他成功项也会回滚吗

不能把多主题订阅当成应用层事务。建议保留逐项结果，并由调用方明确选择部分运行还是清理已建立订阅。

等待超时后能立即无限重试吗

不宜。超时不等于请求取消。应先保留不确定状态，结合后续Token结果和连接状态安排有上限的恢复，避免不断叠加同类请求。

参考：Paho Client API；Paho Token实现；OASIS MQTT 3.1.1规范。