工程学习台

主题

主要章节

设备身份与 MQTT 会话：连接、QoS、订阅与遗嘱 ​

一台设备上电后做的第一件事不是发数据，而是证明"我是谁"，然后申请一条能被 broker 记住的连接。这两步决定了后面所有事情的含义：消息投给谁、断线后是否补发、重复投递要不要去重、掉线时谁替设备说话。本页把 MQTT 5 的设备接入拆成身份、会话、QoS、订阅与保留/遗嘱四组结论，每一组都用真实嵌入式 HiveMQ Community Edition broker 与真实 MQTT 5 客户端重新跑过一遍，不是对着配置清单推演。

本页是 U88（#125）的 D-IOT 权威正文：设备身份/ACL、连接与会话、QoS 语义、订阅/保留/遗嘱，以及"QoS 不推出业务 exactly-once"这一条最容易在面试里答错的边界。设备影子、命令状态机与跨语言契约是 U89 的范围，本页不展开；投递语义的一般模型（至少一次、至多一次、业务幂等）归MQ 生产存储消费可靠性，业务去重状态机归重复消费、幂等与顺序消息，本页只讲 MQTT 协议层与它们的分界线。TLS 握手的细节归HTTP 与 TLS，网关侧的身份传递归 Gateway 可靠性。

设备接入的四个问题 ​

把"设备连上来发数据"拆开，是四个方向不同、失败方式不同的问题：

问题协议位置答错时的后果本页章节
这台设备是谁，能操作哪些主题CONNECT 凭证 + 扩展点认证 + TopicPermission仿冒设备发布伪数据，或一台设备读到全厂数据设备身份、主题级 ACL
断开再连上，还是不是同一条会话CONNECT 的 cleanStart / sessionExpiryInterval重启后订阅丢失、命令没收到，或旧会话无限堆积会话
一条消息投几次、什么时候算送达PUBLISH 的 QoS 与确认报文把"至少一次"当成"恰好一次"，业务重复扣款QoS 与确认
新订阅者如何拿到最新状态，掉线如何被发现retain 位、Will 字段面板显示过期值；设备断线十分钟无人知道订阅、保留与遗嘱

这四个问题共享一个前提：broker 只对自己承诺的语义负责。设备身份是"谁能连"，ACL 是"连上后能碰什么"，会话是"断线后还记得什么"，QoS 是"每条消息确认到哪一步"。任何一条被省略，后面都会被误读成别的保证。

版本、工程与观察入口 ​

固定 HiveMQ Community Edition 2026.5（commit fcc2244fa3ee9c96dc1c57bda2679328a47756e0，Apache-2.0）、安装其官方扩展点所需的 HiveMQ Extension SDK 4.52.0（CE 2026.5 的 POM 声明版本，commit fb5ae38836c392e0813c2a99b20da7275c2f58c8）、客户端 HiveMQ MQTT Client 1.4.1（tag v1.4.1，commit 43a3105c9ccbf1fd79aaae4f18dbdf713cf6f1e1）。实际解析到的 Netty 为 4.1.134.Final。运行环境是 macOS 27.2 arm64、BellSoft Liberica JDK 21.0.11+11-LTS、Maven 3.9.9。查询与运行日期 2026-10-27。

选型先说明白：同样可以嵌入 JVM 的 Moquette，Maven Central 上最新发布停在 0.17（2023 年），GitHub 已推进到 0.18.5 但 artifact 未同步发布，且只支持 MQTT 3.x；HiveMQ CE 同时支持 MQTT 3.x 与 5，通过 hivemq-community-edition-embedded 提供嵌入式入口，并有官方扩展点做认证与 ACL。因此本页以 CE 为主。选它不是唯一正确答案，而是"协议版本 + 认证扩展点 + 可嵌入"三个条件同时满足的最短路径；正文给出的结论只依赖 MQTT 5 协议语义与 broker 类行为，不依赖某一家的实现细节。

公开入口是一条命令：

实验代码sh
python3 labs/iot-device/run.py

run.py 使用 JDK 21 执行 labs/iot-device 的 clean test，核对两份 Surefire 报告共 15 个测试，解析依赖版本，校验 15 条 U88 观察行，把输入、退出码、版本、观察行与文件 SHA-256 写入 labs/iot-device/evidence/u88-2026-10-27.json。每个测试自己启动一个嵌入式 broker：随机端口监听 127.0.0.1、独立的临时配置与数据目录、真实 Netty TCP 监听器、file 持久化模式；测试结束 stop+close 并删除临时目录。没有 listener mock，没有断言内部字段。

本页的框架原码均来自固定 tag 的连续片段，元数据带整文件 SHA-256 与行范围；发布源码包与官方仓库逐字节一致（已核）。教学扩展、客户端配置与运行观察另行标注。源码、配置、编译、真实连接是四种不同强度的证据，互不替代。

设备身份：三个载体 ​

传输层、协议层与应用层 ​

"设备身份"至少有三种承载方式，经常被混为一谈：

载体证明什么典型实现不能证明什么
传输层：TLS 客户端证书持有私钥的设备mTLS，证书 CN 用作 clientId设备内部逻辑没被篡改
协议层：CONNECT 的凭证知道 username/password 或能完成 AUTH 交换MQTT 5 用户名密码、增强认证（AUTH 报文）凭证不是从另一台设备复制来的
应用层：业务注册表这台设备在平台登记过、属于某个租户一机一密、设备三元组、动态注册传输通道是加密的

本实验用第二种：clientId、username、password 三者由设备 id 派生，broker 扩展点校验一致性。这是教学上用最短路径展示"身份决定权限归属"的模型；生产里常见的是三者叠加——mTLS 保证通道与设备一一对应，CONNECT 凭证再绑业务租户，注册表决定这台设备能不能上线。三者叠加不冗余，因为各自防的攻击不同：证书防冒名连接，凭证防复制设备镜像，注册表防"设备合法但已报废"。

常见错误答案：把 clientId 当作身份。clientId 只是一个字符串，任何客户端都能声明 clientId=dev-a；它的作用是"占住这个会话"，不是"证明自己是 dev-a"。本实验专门验证了这一点：谁能提供正确凭证，谁就能用同一 clientId 接管会话（见重连裁决与 takeover）。

CONNECT 与 CONNACK：身份协商 ​

MQTT 5 的 CONNECT 把身份信息放在三个位置：payload 里的 clientId（必有）、username、password，以及 properties 里的 session expiry interval 等会话参数。broker 的答复是 CONNACK，两个关键字段：

  • reason code：0x00 成功；0x86 bad user name or password；0x87 not authorized。身份失败与权限失败是不同码，运维排障时应先看码再猜原因。
  • session present 位：本次连接是否复用了已有会话。它是判断"设备重连后能不能收到离线命令"的唯一协议依据，不是"连接成功"的同义词。

客户端侧这两个参数必须显式给出，不能依赖默认值。HiveMQ MQTT Client 的构建器文档就写明了取值范围与语义：

原码java
     * Sets whether the client {@link Mqtt5Connect#isCleanStart() starts a clean session}.
     *
     * @param cleanStart whether the client starts a clean session.
     * @return the builder.
     */
    @CheckReturnValue
    @NotNull B cleanStart(boolean cleanStart);

    /**
     * Sets the {@link Mqtt5Connect#getSessionExpiryInterval() session expiry interval} in seconds.
     * <p>
     * The value must be in the range of an unsigned int: [0, 4_294_967_295]. 4_294_967_295 disables session expiry.
     *
     * @param sessionExpiryInterval the session expiry interval in seconds.
     * @return the builder.
     */
    @CheckReturnValue
    @NotNull B sessionExpiryInterval(long sessionExpiryInterval);

两个方法签名背后是两个独立维度：cleanStart 决定"这次连接是否丢弃旧会话"，sessionExpiryInterval 决定"这次断开后会话保留多久"。常见错误是只设一个：cleanStart=false 但不设 expiry，等于用默认值（MQTT 5 里是 0，即断线即删）；cleanStart=true 但设了 expiry，等于每次重连都丢弃订阅，expiry 形同虚设。实验验证了三种真实组合：cleanStart=false 首次连接 sessionPresent=false，cleanStart=false 重连 sessionPresent=true，cleanStart=true 连接 sessionPresent=false；按协议 cleanStart=true 时 session present 必须为 0，"cleanStart=true 且 sessionPresent=true" 不存在。

认证扩展点的真实调用链 ​

HiveMQ CE 的嵌入式入口有一个必须先知道的默认值：

原码java
        log.info("Setting default authentication behavior to ALLOW ALL");
        InternalConfigurations.AUTH_DENY_UNAUTHENTICATED_CONNECTIONS.set(false);

嵌入式模式把"拒绝未认证连接"的开关关掉了——这是为了让开发者先跑通，但意味着没有安装认证扩展时，任何人都能匿名连接。本实验专门跑了一个反例：不装扩展的 broker 上，匿名客户端 CONNACK 0x00 成功。生产里如果只看到"broker 启动成功"就上线，等于把这条默认值当成了安全配置。

再看安装了认证扩展之后的路径。connectSuccessfulUndecided 是"没有任何扩展给出明确结论"的分支；如果安全开关是开的，它会回 NOT_AUTHORIZED 并断开：

原码java
    public void connectSuccessfulUndecided(
            final @NotNull ChannelHandlerContext ctx,
            final @NotNull ClientConnectionContext clientConnectionContext,
            final @NotNull CONNECT connect,
            final @Nullable ModifiableClientSettingsImpl clientSettings) {
        clearPasswordIfWanted(clientConnectionContext);
        if (AUTH_DENY_UNAUTHENTICATED_CONNECTIONS.get()) {
            mqttConnacker.connackError(clientConnectionContext.getChannel(),
                    PluginAuthenticatorServiceImpl.AUTH_FAILED_LOG,
                    ReasonStrings.AUTH_FAILED_NO_AUTHENTICATOR,
                    Mqtt5ConnAckReasonCode.NOT_AUTHORIZED,
                    ReasonStrings.AUTH_FAILED_NO_AUTHENTICATOR,
                    Mqtt5UserProperties.NO_USER_PROPERTIES,
                    true);
            return;
        }
        clientConnectionContext.proposeClientState(ClientState.AUTHENTICATED);

注意 connectSuccessfulUndecided 的名字：它不是"认证成功"，是"没有认证器表态"。扩展必须显式调用 authenticateSuccessfully() 或 failAuthentication()；两个都不调用，在嵌入式默认配置下会落到 proposeClientState(AUTHENTICATED)，在安全开关打开时会被拒绝。这就是为什么"装了扩展"不等于"认证生效"——要看扩展是否真的注册了 provider，以及它是否在每条路径上都给了结论。

实验里的扩展是真实的 broker 扩展点装配，不是测试桩：Services.securityRegistry().setAuthenticatorProvider(...) 注册一个 SimpleAuthenticator，在 onConnect 里校验 username == "device-" + clientId 与 password == "secret-" + clientId，失败走 failAuthentication(ConnackReasonCode.NOT_AUTHORIZED)，成功则向 ModifiableDefaultPermissions 添加一条 TopicPermission 后 authenticateSuccessfully()。这条链路的每一环都在 CE 的公开扩展 API 上，源码版本固定为 2026.5 与 SDK 4.52.0。

主题级 ACL：一条规则挡在哪里 ​

TopicPermission 的五个维度 ​

认证只回答"你是设备 dev-a"，ACL 回答"dev-a 能碰哪些主题的哪些操作"。HiveMQ 的 TopicPermission 把一条规则拆成五个维度：

维度取值作用
topicFilter任意带通配符的主题过滤器规则匹配范围
typeALLOW / DENY命中后放行还是拒绝
activityPUBLISH / SUBSCRIBE / ALL发布与订阅分别授权
qosZERO / ONE / TWO / ZERO_ONE … ALL该规则允许的最大 QoS
retainRETAINED / NOT_RETAINED / ALL是否允许保留消息
sharedSubscription / sharedGroupSHARED / NOT_SHARED / ALL，# 表示任意组共享订阅的授权粒度

本实验给设备只授一条规则：devices/<clientId>/#，activity=ALL、qos=ALL、retain=ALL。svc- 前缀的服务账号授 devices/#。这个"设备只能碰自己的子树"是工业 IoT 最常见的最小授权模型；不同租户再加一层前缀（tenants/<tid>/devices/<id>/#）即可。

默认行为与越权表现 ​

规则的默认值决定了兜底方向。HiveMQ 的发布检查在没有任何规则时是 DENY，有规则但都不命中时按默认行为走：

原码java
    private static boolean checkPublish(
            final @Nullable ModifiableDefaultPermissions permissions,
            final @NotNull String topic,
            final @NotNull QoS qos,
            final boolean retain) {
        if (permissions == null) {
            // no permissions set -> default to DENY
            return false;
        }
        final List<TopicPermission> topicPermissions = permissions.asList();
        if (topicPermissions.size() < 1) {
            return permissions.getDefaultBehaviour() == DefaultAuthorizationBehaviour.ALLOW;
        }
        final String[] splitTopic = StringUtils.splitPreserveAllTokens(topic, "/");
        final String stripedTopic;
        if (topic.length() > 1) {
            stripedTopic = StringUtils.stripEnd(topic, "/");
        } else {
            stripedTopic = topic;
        }
        for (final TopicPermission topicPermission : permissions.asList()) {
            if (implied(topicPermission, stripedTopic, splitTopic, qos, TopicPermission.MqttActivity.PUBLISH, retain)) {
                return topicPermission.getType() == TopicPermission.PermissionType.ALLOW;
            }
        }
        return permissions.getDefaultBehaviour() == DefaultAuthorizationBehaviour.ALLOW;
    }

三个细节值得记住。第一，permissions == null 时直接 DENY——"忘了设置权限"导致的是拒绝而不是放行，方向是对的。第二，一旦添加了第一条 ALLOW 规则，ModifiableDefaultPermissionsImpl.add 会把默认行为从 ALLOW 翻成 DENY（源码见 ModifiableDefaultPermissionsImpl，本页未摘录，只作说明）；所以"只加了一条允许规则"等于"其余全部拒绝"。第三，匹配是第一条命中即返回，ALLOW 与 DENY 的顺序会影响结果——这是规则引擎的常见坑，不要把互相矛盾的规则写在一起。

越权行为在不同操作上的表现完全不同，这是排障时最容易误判的地方。

订阅越权：SUBACK 返回 NOT_AUTHORIZED。订阅被拒但连接保留：

原码java
            // then we deny the subscription
            if (authorizersPresent && (defaultPermissions == null || (defaultPermissions.asList().size() < 1 &&
                    defaultPermissions.getDefaultBehaviour() == DefaultAuthorizationBehaviour.ALLOW &&
                    !defaultPermissions.isDefaultAuthorizationBehaviourOverridden()))) {
                providedCodes[i] = Mqtt5SubAckReasonCode.NOT_AUTHORIZED;
                continue;
            }
            final Topic subscription = msg.getTopics().get(i);
            if (!DefaultPermissionsEvaluator.checkSubscription(permissions, subscription)) {
                providedCodes[i] = Mqtt5SubAckReasonCode.NOT_AUTHORIZED;
                // build reason string (concat multiple reasons)
                reasonStrings[i] = "Not authorized to subscribe to topic '" + subscription.getTopic() + "' with QoS '" +
                        subscription.getQoS().getQosNumber() + "'";
            }

发布越权：MQTT 5 先回错误 PUBACK（或 PUBREC），再断开；MQTT 3.1.1 直接断开、没有回执。QoS0 因为本来就没有回执通道，只被静默丢弃——设备完全无法感知：

原码java
    private void finishUnauthorizedPublish(
            @NotNull final ChannelHandlerContext ctx,
            @NotNull final PUBLISH publish,
            @Nullable final AckReasonCode reasonCode,
            @Nullable final String reasonString) {
        final ClientConnection clientConnection = ClientConnection.of(ctx.channel());
        clientConnection.setIncomingPublishesDefaultFailedSkipRest(true);
        if (!ctx.channel().isActive()) {
            // no more processing needed.
            return;
        }
        final String reason = "Not authorized to publish on topic '" + publish.getTopic() + "' with QoS '" +
                publish.getQoS().getQosNumber() + "' and retain '" + publish.isRetain() + "'";
        // MQTT 3.x.x -> disconnect (without publish answer packet)
        if (clientConnection.getProtocolVersion() != ProtocolVersion.MQTTv5) {
            final String clientId = clientConnection.getClientId();
            mqttServerDisconnector.disconnect(ctx.channel(),
                    "Client '" + clientId + "' (IP: {}) is not authorized to publish on topic '" + publish.getTopic() +
                            "' with QoS '" + publish.getQoS().getQosNumber() + "' and retain '" + publish.isRetain() +
                            "'. Disconnecting client.",
                    reason,
                    Mqtt5DisconnectReasonCode.NOT_AUTHORIZED,
                    reason);
            return;
        }
        // MQTT 5 -> send ACK with error code and then disconnect
        switch (publish.getQoS()) {
            case AT_MOST_ONCE :
                // just drop the message, no back channel to the client
                break;
            case AT_LEAST_ONCE :
                final PUBACK puback = new PUBACK(publish.getPacketIdentifier(),
                        reasonCode != null ? Mqtt5PubAckReasonCode.from(reasonCode) :
                                Mqtt5PubAckReasonCode.NOT_AUTHORIZED,
                        reasonString != null ? reasonString : reason,
                        Mqtt5UserProperties.NO_USER_PROPERTIES);
                ctx.pipeline().writeAndFlush(puback);
                break;
            case EXACTLY_ONCE :
                final PUBREC pubrec = new PUBREC(publish.getPacketIdentifier(),

（摘录第 160–199 行在 case EXACTLY_ONCE 的 PUBREC 构造处截断，后续分支同一模式：回错误码再断开。）

实验验证了这条链：设备用 QoS1 向 devices/dev-b/cmd 发布，收到的 PUBACK reason code 是 0x87，随后 broker 发 DISCONNECT 0x87。结论：越权发布的后果是"QoS1/2 有回执、QoS0 静默丢弃"，所以设备上云的第一条自检不能只发 QoS0 的空数据包——它可能一直失败而你什么都看不到。

连接与会话 ​

会话里存了什么 ​

MQTT 的"会话"是 broker 侧为某个 clientId 保存的一组状态。它不是"连接"，连接断了会话可以还在：

状态cleanStart=true 时cleanStart=false 且未过期时何时清除
订阅关系不继承恢复cleanStart、会话过期、显式 UNSUBSCRIBE
未确认的 QoS1/2（inflight）不继承恢复并标记 DUP 重投收到 PUBACK/PUBCOMP
离线期间新到的 QoS1/2不排队进入会话队列投递并确认，或队列溢出
离线期间的 QoS0丢弃丢弃从不排队
遗嘱（will）每次 CONNECT 重新声明同一连接内有效正常断开后作废

这张表是本页后面所有结论的地基。设备侧最常见的两个 bug 都能从它读出来：一是"重连后收不到命令"，原因是 cleanStart=true（每次都是新会话，离线消息没有地方可存）；二是"重启后收到几天前的旧命令"，原因是 sessionExpiryInterval 设得太大且没有过期清理，离线队列没有上限。

重连裁决与 takeover ​

broker 在收到 CONNECT 时要判断三件事：有没有旧会话、旧会话是否过期、本次是否要求 cleanStart。CE 的实现把这三件事写在一个地方：

原码java
                final ClientSession previousClientSession = connectResult.getPreviousClientSession();
                // Send any pending will if requesting a clean start.
                // As per the specification, a clean start "ends" any existing session and requires its will to be sent,
                // if available.
                // Therefore, immediately send any pending will that was created during the preceding disconnect.
                // If no clean start is required, a client might be re-connecting to the existing session.
                // In this case, cancel any pending send of a will.
                if (cleanStart) {
                    pendingWillMessages.sendWillIfPending(client, previousClientSession);
                } else {
                    pendingWillMessages.cancelWillIfPending(client);
                }
                // CleanUp the client session if the session is expired OR the client is clean start client.
                final ListenableFuture<Void> cleanupFuture;
                if (cleanStart) {
                    cleanupFuture = cleanClientData(client);
                } else {
                    final boolean expired = previousTimestamp != null &&
                            previousClientSession.isExpired(System.currentTimeMillis() - previousTimestamp);
                    if (expired) {
                        // timestamp in milliseconds + session expiry in seconds * 1000 = milliseconds
                        eventLog.clientSessionExpired(
                                previousTimestamp + previousClientSession.getSessionExpiryIntervalSec() * 1000,
                                client);
                        cleanupFuture = cleanClientData(client);
                    } else {

两个分支值得逐字读。cleanStart=true：如果旧会话有未发送的遗嘱，先发遗嘱再清会话——因为 MQTT 5 规范规定 clean start 会结束旧会话，旧会话的遗嘱必须发出。cleanStart=false：如果会话未过期，取消未发遗嘱并复用；如果已过期，按"过期时间 = 上次断开时刻 + expiry 秒"清理。

takeover 是另一个必知行为：同一个 clientId 已有活跃连接时，新连接建立会断开旧连接。实验验证：设备 A 先连上并订阅，设备 B 用同一 clientId 重连（sessionPresent=true），A 被踢下线且收不到后续消息，B 收到命令。这解释了两类线上现象：设备固件重试逻辑反复重连时，旧连接会不断被新连接踢掉，表现为"设备在线但数据时断时续"；反过来，攻击者拿到凭证后能用同一 clientId 顶掉真实设备——所以clientId 撞车本身就是安全事件，监控应以 alarm 而不是 warn 处理。

离线队列的 QoS 边界 ​

离线队列只对 QoS1/2 生效。实验的对照非常直接：设备离线期间服务端发 QoS0 与 QoS1 各一条，设备以 cleanStart=false 重连后只收到 QoS1 那条，QoS0 永久丢失。这不是 broker 的 bug，是 QoS0 的定义——它没有确认，broker 没有任何义务保存。

队列长度是另一个容量约束。CE 的 mqtt.queued-messages.max-queue-size 与 strategy 决定队列满时的行为（丢弃最旧/丢弃最新/拒绝），本实验使用默认值，未做溢出压测；正文只能给出"必须显式配置队列上限并观测溢出计数"的结论，不能给出具体容量数字。这是 K08 的边界：负载、窗口与不变量要可计算，但本页没有跑出生产容量。

Keep-alive 与掉线检测 ​

keep-alive 是 CONNECT 时双方协商的秒数（0 表示关闭）。协议规定：客户端承诺在 keep-alive 内至少发一条控制报文（PINGREQ 也算）；broker 在 1.5 倍 keep-alive 内没有收到任何报文，就判定连接失效，关闭连接并触发 will。这个机制解决的是 TCP 自身发现不了的"半开连接"：设备断电、进隧道、NAT 表项过期时，一端以为连接还在，另一端的 socket 仍然读写正常，只有应用层心跳能发现。

三个工程约束。第一，keep-alive 是双向承诺，不是 broker 单方面的超时；客户端设置的 60 秒与 broker 允许的最大值之间取较小者。第二，keep-alive 的粒度决定了掉线发现的延迟下界：60 秒 keep-alive 意味着最坏 90 秒后 broker 才知道设备掉线，平台侧的状态展示必须接受这个窗口，不能显示"实时在线"。第三，keep-alive 与 will 联动：will 的触发条件之一就是 keep-alive 超时，所以"设备离线多久后同事收到告警"约等于 1.5 倍 keep-alive + 告警链路延迟。

本实验未实跑 keep-alive 超时路径：等待 1.5 倍 keep-alive 且稳定制造半开 TCP 需要额外环境控制，正文只给规范依据与源码路径（will 触发在 ClientSessionPersistenceImpl 与连接关闭路径中），不宣称已运行。

QoS：握手、重投与去重 ​

三种 QoS 的数据流 ​

QoS报文序列谁发保证成本
0PUBLISH客户端→broker，broker→订阅者最多一次，可能丢最低，无状态
1PUBLISH → PUBACK接收方回 PUBACK至少一次，可能重复需要 packet id 与 inflight 状态
2PUBLISH → PUBREC → PUBREL → PUBCOMP双向四次不重复（对 broker 单次投递）两次往返 + 持久化

实验用原始 MQTT 5 报文探针跑完整个 QoS2 握手：发 PUBLISH(QoS2, id=11) → 收到 PUBREC → 发 PUBREL → 收到 PUBCOMP，订阅方应用层只收到一次。这一条结论很精确：**QoS2 保证的是"broker 不会把同一条消息向订阅者投递两次"，前提是握手完整、同一个 broker、会话状态没丢。**它不保证业务只执行一次（见下一节）。

DUP 是提示，不是去重凭据 ​

重投时 broker 必须设置 DUP 位。CE 在从持久化队列读出 inflight 消息时把 duplicateDelivery 置为 true：

原码java
                    final MessageWithID message = serializer.deserializeValue(serializedValue);
                    // This works because in-flight messages are always first in the queue
                    if (message.getPacketIdentifier() == ClientQueuePersistenceSerializer.NO_PACKET_ID) {
                        return false;
                    }
                    if (message instanceof PUBLISH) {
                        final PUBLISH publish = (PUBLISH) message;
                        if (!setPayloadIfExistingElseDrop(publish, client, shared, bucketIndex)) {
                            return true;
                        }
                        bytes[0] += publish.getEstimatedSizeInMemory();
                        publish.setDuplicateDelivery(true);
                    }
                    messages.add(message);
                    count[0]++;

编码器再把它写进固定头的第 3 位：

原码java

    private static void encodeFixedHeader(final @NotNull PUBLISH publish, final @NotNull ByteBuf out) {
        int flags = 0;
        if (publish.isDuplicateDelivery()) {
            flags |= 0b1000;
        }
        flags |= publish.getQoS().ordinal() << 1;
        if (publish.isRetain()) {
            flags |= 0b0001;
        }
        out.writeByte(FIXED_HEADER | flags);
        MqttVariableByteInteger.encode(publish.getRemainingLength(), out);
    }

实验用原始报文完整验证了这条链：订阅方收到第一条 QoS1（DUP=0）后不确认、直接关闭 socket；以 cleanStart=false 重连，同一 packet id 的重投报文 DUP=1。

客户端侧也把 DUP 当作协议不变量来检查——收到"看起来是重投但没设 DUP"的报文，HiveMQ 客户端直接以协议错误断开：

原码java
    private boolean checkDupFlagSet(
            final @NotNull ChannelHandlerContext ctx, final @NotNull MqttStatefulPublish publish) {

        if (publish.isDup()) {
            return true;
        }
        LOGGER.error("DUP flag must be set for a resent PUBLISH ({})", publish);
        MqttDisconnectUtil.disconnect(ctx.channel(), Mqtt5DisconnectReasonCode.PROTOCOL_ERROR,
                "DUP flag must be set for a resent QoS " + publish.stateless().getQos().getCode() + " PUBLISH");
        return false;
    }

关键推论要说得比"重复看 DUP"精确。规范 [MQTT-3.3.1-1] 要求：同一 PUBLISH 报文被重投时，发送方必须置 DUP=1；而且进入的 DUP 值不会被传播，broker 发给订阅者的报文 DUP 由它自己决定。所以"协议层的同一报文重投"确实带 DUP=1。但业务意义上的重复远不止这一种：

  1. 应用层重发：设备超时后重新构造一条 PUBLISH（新 packet id），DUP=0，内容却相同。
  2. 重新连接后重放：设备把未确认的业务指令存在本地，重连后重新发布，DUP=0。
  3. broker 崩溃/桥接/集群迁移：跨实体的状态与去重不共享，对端可能收到两次投递，其中一次是"首次发送"而非协议重投。
  4. QoS2 的重复发生在 PUBREL 阶段，应用看到的 PUBLISH 只出现一次，但业务指令可能已被重复执行。

因此 DUP 只能回答"这条是不是协议层的重投"，不能回答"这个业务动作是不是已经发生过"。业务去重必须靠业务幂等键（设备 id + 指令 id + 序号/版本），不能靠 DUP 位。本实验只验证了第 1 类以外的场景中"协议重投置 DUP=1"这一条，其余三类按规范与机制说明，未逐类实跑。

投递 QoS 降级 ​

实际投递 QoS 是发布 QoS 与订阅 QoS 的较小值。实验：设备以 QoS0 订阅，服务端用 QoS2 发布，设备收到的是 QoS0。这条规则解释了为什么"全链路 QoS2"必须两端都声明 QoS2：任何一跳降级，后面所有保证都降级。常见错误答案：只把发布端设成 QoS2 就认为"现在不会重复了"——订阅端的 QoS 才是投递强度上限。

订阅、保留与遗嘱 ​

主题过滤器与订阅 ​

MQTT 主题是 / 分隔的层级字符串，订阅用通配符：

  • + 匹配一层；devices/+/telemetry 匹配 devices/a/telemetry，不匹配 devices/a/b/telemetry。
  • # 匹配多层且必须在末尾；devices/# 匹配 devices/a、devices/a/b，也匹配 devices 本身（带 # 的父层）。
  • 以 $ 开头的主题（$SYS、$share）不被 # 或 + 匹配，必须显式写。
  • 发布者不需要知道订阅者；订阅者用通配符聚合多个设备的数据。

ACL 也是按 topic filter 匹配的，所以订阅通配符的粒度就是数据暴露的粒度：devices/# 给运维服务可以，给单台设备就等于把全厂数据交出去。这是 A04 与 D01 的交点。

保留消息的生命周期 ​

保留消息是 broker 为"最新状态"提供的机制，语义只有四条：

  1. 发布时 retain=true → broker 把这条消息存为该主题的保留消息（覆盖旧的）。
  2. 新订阅者订阅该主题（过滤器命中）→ 立即收到保留消息，retain 位在投递时为 1。
  3. 清除：向该主题发布 retain=true 且 payload 为空 → broker 删除保留消息。
  4. 持久化：file 持久化模式下保留消息随 broker 正常重启存活。

实验逐条验证前三条：发布 online（retain=true）→ 新订阅者立刻收到；发空 payload（retain=true）→ 新订阅者收不到（清除后订阅收不到任何东西）。

保留消息与 broker 重启 ​

第四条"持久化"需要单独说，因为它最容易被过度外推。file 持久化模式下，保留消息写入 broker 的持久化目录，正常 stop+close 后再启动，新订阅者仍收到 persisted——实验验证了这条。但正常 stop 后重开 ≠ kill -9 或断电：正常 stop 保证目录被干净刷盘并关闭；kill -9 或断电取决于写入时机、文件系统缓存与恢复过程，本实验未验证。设备影子（U89）会建立在保留消息之上，所以这条边界必须在 U88 说清：影子可以依赖"保留消息跨正常重启存活"，不能依赖"跨崩溃一致"。

遗嘱：掉线的最后一条消息 ​

遗嘱（Will）是设备在 CONNECT 时声明的一条消息，broker 在非正常断开时代为发布。触发条件精确地说是：

断开方式will 是否发布
客户端发 DISCONNECT（正常断开）不发布
TCP 连接被关闭/重置，未发 DISCONNECT发布
keep-alive 超时（1.5 × keep-alive 内无报文）发布
协议错误被 broker 断开发布
同 clientId 的新连接接管，且旧连接 Will Delay 为 0旧连接的 will 发布
同 clientId 的新连接指定 cleanStart=false 且旧连接 Will Delay>0，并在到期前到达will 被取消，不发布
同 clientId 的新连接指定 cleanStart=truewill 发布（会话结束触发），与 Will Delay 无关
cleanStart=true 重连且旧会话有暂存 will先发 will 再清会话

Will Delay Interval 是 MQTT 5 新增的 will 属性（CONNECT 的 Will Properties 里）。规范 [MQTT-3.1.2-8] 的表述是：连接关闭后，要等 Will Delay 到期或会话结束才发布 will。规范 3.1.3.2.2 的非规范性说明进一步给出接管场景：新连接使用已存在连接的 clientId 时，旧连接的 will 会发送，除非新连接指定 cleanStart=0 且 Will Delay>0；如果 Will Delay 为 0，will 在旧连接关闭时立即发送；如果 cleanStart=1，will 因会话结束而发送。默认 Will Delay 为 0，所以"掉线立刻发 will"。这个属性解决的是设备快速重连的场景：网络抖动导致断开又立刻重连时，业务不希望看到"离线"再"在线"的抖动。本实验没有设置 Will Delay，跑的是默认 0 的行为。

规范 [MQTT-3.1.2-5] 明确列出的触发情形包括：Server 检测到 I/O 错误或网络故障、客户端在 Keep Alive 内没有通信、客户端未发 DISCONNECT（Reason Code 0x00）就关闭连接、Server 未收到 DISCONNECT（0x00）就关闭连接。

实验验证了两种：原始报文探针建连（带 will）后直接关闭 socket，不发送 DISCONNECT，订阅服务收到 offline；用 HiveMQ 客户端正常 disconnect()，服务收不到 will。本实验未跑 keep-alive 超时路径（要等 1.5 × keep-alive，且需要半开 TCP 才能稳定复现），正文只能给源码/规范依据，不能宣称已运行。

遗嘱的工程价值在于"掉线是常态"：设备可能断电、可能进隧道、可能固件崩溃，任何一种都不会发 DISCONNECT。没有 will，平台只能等 keep-alive 超时才知道设备离线；有 will，订阅者能立刻收到一条明确的"我下线了"。但 will 不是心跳——它只在断开时发一次，不能替代周期性上报。

QoS 不推出业务 exactly-once ​

这是本页最重要的边界，也是面试里最容易失分的地方。

QoS2 保证的是：在一条会话、一个 broker、握手完整的前提下，broker 不会把同一条 PUBLISH 向订阅者投递两次。

QoS2 不保证的是：

场景QoS2 是否仍然成立业务后果
broker 正常重启，会话未过期成立（inflight 持久化）无重复
broker 崩溃后从持久化恢复可能重复（崩溃时 inflight 状态可能不一致）业务可能执行两次
跨 broker 桥接/集群迁移不保证（不同 broker 的去重状态不共享）可能重复
订阅者应用处理完但未持久化就崩溃不保证（应用重启后会再收一次）业务可能执行两次
业务副作用跨越多个系统（扣款+发命令）完全不相关需要业务事务/幂等

结论的准确表述：QoS 是"传输层对消息传递次数的承诺"，业务 exactly-once 是"业务状态机对效果次数的承诺"。前者最多做到"broker 到订阅者一次"，后者必须靠幂等键、去重表、状态查询与对账。这条边界在 MQ 生产存储消费可靠性 里有 Kafka 侧的对应结论（幂等 producer 去重依赖 epoch 与 sequence，事务有超时与 abort 边界），在重复消费、幂等与顺序消息 里有消费者侧的 Inbox 模式——本页只负责把 MQTT 这一段的承诺说准，不重复它们的机制。

可计算的不变量（K08）：设设备每秒发 N 条、每条业务效果需要一次数据库写入，broker 重启概率 p、应用崩溃概率 q。最坏情况下业务侧重复执行次数 ≈ N × (p + q) × 窗口。这个式子只用来比较方案：QoS1 + 幂等表 vs QoS2 + 无幂等，前者在应用崩溃时不比后者差，而网络开销只有一半。真正的选择依据是业务能不能承担重复，不是 QoS 数字高低。

观测、排障与恢复顺序 ​

设备接入故障的定位顺序应该固定，否则会在错误的方向上重启服务：

  1. 看 CONNACK reason code：0x86 是凭证错（检查 username/password/证书），0x87 是权限或未认证（检查扩展是否加载、ACL 规则），0x00 才是连上。
  2. 看 sessionPresent：false 表示没有复用会话（cleanStart=true 或会话已过期），这解释了"为什么没收到离线命令"。
  3. 看 SUBACK reason codes：0x87 表示订阅被 ACL 拒绝，连接还在但收不到数据。
  4. 看 PUBACK reason code：0x87 表示发布被拒；QoS0 没有回执，只能看 broker 日志。
  5. 看 DUP 与 packet id：DUP=1 频繁出现说明网络抖动或订阅者确认慢，不是 broker 在"重复发"。
  6. 看遗嘱是否触发：will 触发说明设备非正常断开；未触发但设备离线，说明 keep-alive 还没到。

恢复顺序也有讲究：先确认身份与 ACL（否则重连会立刻被踢），再确认会话参数（否则订阅会丢），最后才放业务流量。反过来做等于用新错误掩盖旧错误。

本实验的观测能力边界：CE 的指标注册表（getMetricRegistry()）可用，但本实验未导出 Prometheus，也未做告警；上述顺序来自协议语义与实验观察，不是生产监控方案。

实验读法 ​

python3 labs/iot-device/run.py 跑 15 个测试，观察行前缀是 U88：

  1. U88 identity=PASS ...：正确凭证连上，自己的主题订阅 GRANTED_QOS_1，他人主题 SUBACK 0x87。
  2. U88 identityReject=PASS reason=NOT_AUTHORIZED：错误密码 CONNACK 0x87。
  3. U88 anonymousReject=PASS reason=NOT_AUTHORIZED：装了认证扩展后匿名连接被拒。
  4. U88 embeddedDefaultAllowAll=PASS anonymous=SUCCESS：反例——不装扩展的嵌入式 broker 允许匿名连接，这是默认值不是配置错误。
  5. U88 aclPublish=PASS pubackReason=0x87 disconnectReason=0x87：越权 QoS1 发布的协议表现。
  6. U88 session=PASS restored=true offline=[cmd-1, cmd-2, cmd-3]：持久会话恢复并补发离线消息。
  7. U88 cleanStart=PASS sessionPresent=false leak=false：cleanStart 丢弃旧会话，旧订阅不泄漏。
  8. U88 qosOffline=PASS qos0=lost qos1=[kept]：离线队列只保 QoS1/2。
  9. U88 qos2=PASS deliveredOnce=true：QoS2 完整握手，应用只收一次。
  10. U88 dupRedelivery=PASS dup=true：未确认 QoS1 重投设置 DUP。
  11. U88 qosDowngrade=PASS published=2 subscribed=0 delivered=0：投递 QoS 取 min。
  12. U88 takeover=PASS sessionPresent=true oldKicked=true deliveredToNew=...：同 clientId 接管。
  13. U88 retained=PASS before=online afterClear=none：保留消息投递与清除。
  14. U88 retainedRestart=PASS value=persisted：保留消息跨正常重启。
  15. U88 will=PASS abruptWill=offline gracefulWill=false：突然断开触发遗嘱，优雅断开不触发。

证据 JSON 记录了一次完整运行：JDK/Maven/OS 版本、依赖解析版本、两份 Surefire 报告、15 条观察行、每个源文件的 SHA-256、清理说明与未运行清单。完整命令、退出码与计数见运行证据，工程边界见实验说明。

闭卷验收 ​

本页 25 道闭卷题收进了下方自测面板：先回忆，再展开参考答案与评分点，最后给三档自评。

未运行与不能由本页证明的 ​

  • TLS/mTLS 与客户端证书、MQTT 5 增强认证（AUTH 报文）、WebSocket 监听器、外部凭据源（数据库/OAuth）均未运行。
  • keep-alive 超时触发遗嘱、同 clientId 接管时旧连接遗嘱的发布时序、MQTT 3.1.1 的行为差异均只给规范/源码依据，未实跑。
  • broker 集群、桥接、共享订阅跨节点、生产容量与队列溢出压测未运行。
  • broker 重启是正常 stop+close 后重开，不是 kill -9、断电或文件系统损坏恢复。
  • 本页的扩展是教学实现，不是生产设备证书体系或动态注册流程；生产应由 mTLS + 注册表 + 短期凭证组合。
  • 教材完成不等于学习者掌握；未运行条件不得由"命令写好"或"配置正确"推出。

工程学习台　阅读讲解，学后自主自测。记录保存在本机浏览器。